"""散落模块（config / manifest / scanner / sorter / auth）单元测试。 运行: python tests/unit/test_modules.py """ from __future__ import annotations import argparse import io import json import os import sys import tempfile import unittest from unittest.mock import MagicMock, patch sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "assets")) from sapcli.config import SAPConfig, _find_config_file, list_profiles, load_config from sapcli.exceptions import CyclicDependencyError, ConfigError from sapcli.manifest import ( CURRENT_VERSION, MANIFEST_FILENAME, Manifest, ManifestEntry, init_manifest, ) from sapcli.scanner import DIRECTORY_TYPE_MAP, ScannedObject, scan_project from sapcli.sorter import TYPE_PRIORITY, topological_sort # ═══════════════════════════════════════════════════════════════════ # config.py # ═══════════════════════════════════════════════════════════════════ class TestSAPConfig(unittest.TestCase): """SAPConfig dataclass 创建与校验。""" def test_create_valid(self): cfg = SAPConfig(host="myhost", client="100", user="USER1", password="pass") self.assertEqual(cfg.host, "myhost") self.assertEqual(cfg.client, "100") self.assertEqual(cfg.user, "USER1") self.assertEqual(cfg.password, "pass") def test_frozen(self): cfg = SAPConfig(host="h", client="100", user="u", password="p") with self.assertRaises(AttributeError): cfg.host = "other" # type: ignore[misc] def test_missing_host_raises(self): with self.assertRaises(ConfigError): SAPConfig(host="", client="100", user="u", password="p") def test_missing_user_raises(self): with self.assertRaises(ConfigError): SAPConfig(host="h", client="100", user="", password="p") def test_missing_password_raises(self): with self.assertRaises(ConfigError): SAPConfig(host="h", client="100", user="u", password="") def test_missing_multiple_raises(self): with self.assertRaises(ConfigError) as ctx: SAPConfig(host="", client="100", user="", password="") msg = str(ctx.exception) self.assertIn("host", msg) self.assertIn("user", msg) class TestFindConfigFile(unittest.TestCase): """_find_config_file 搜索路径逻辑。""" def test_returns_none_when_no_file(self): with tempfile.TemporaryDirectory() as tmp: with patch("os.getcwd", return_value=tmp): result = _find_config_file() if result is not None: self.assertTrue(os.path.isfile(result)) def test_finds_file_in_cwd(self): with tempfile.TemporaryDirectory() as tmp: cfg_path = os.path.join(tmp, "config.ini") with open(cfg_path, "w") as f: f.write("[SAP]\nhost=example.com\n") with patch("os.getcwd", return_value=tmp): result = _find_config_file() self.assertIsNotNone(result) self.assertEqual(result, cfg_path) def _write_tmp_ini(content: str) -> str: """写临时 .ini 文件，返回路径（文件已关闭，可在 Windows 上删除）。""" fd, path = tempfile.mkstemp(suffix=".ini") with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(content) return path class TestListProfiles(unittest.TestCase): """list_profiles 从 config.ini 读取 sections。""" def test_reads_sections(self): path = _write_tmp_ini("[SAP]\nhost=h1\n[DEV]\nhost=h2\n[PRD]\nhost=h3\n") try: profiles = list_profiles(config_path=path) self.assertIn("SAP", profiles) self.assertIn("DEV", profiles) self.assertIn("PRD", profiles) self.assertEqual(len(profiles), 3) finally: os.unlink(path) def test_empty_file_returns_empty(self): path = _write_tmp_ini("") try: profiles = list_profiles(config_path=path) self.assertEqual(profiles, []) finally: os.unlink(path) def test_no_config_path_no_file(self): with patch("sapcli.config._find_config_file", return_value=None): profiles = list_profiles(config_path=None) self.assertEqual(profiles, []) class TestLoadConfig(unittest.TestCase): """load_config 从文件和环境变量读取。""" @staticmethod def _write_config(path: str, section: str = "SAP", **kwargs: str) -> None: import configparser cfg = configparser.ConfigParser() cfg[section] = kwargs with open(path, "w", encoding="utf-8") as f: cfg.write(f) def _clean_env(self): return {k: v for k, v in os.environ.items() if k not in ("SAP_HOST", "SAP_CLIENT", "SAP_USER", "SAP_PASSWORD")} def test_load_from_file(self): fd, path = tempfile.mkstemp(suffix=".ini") os.close(fd) try: self._write_config(path, host="myhost.com", client="200", user="MYUSER", password="secret") with patch.dict(os.environ, self._clean_env(), clear=True): cfg, loaded_from = load_config(config_path=path) self.assertEqual(cfg.host, "myhost.com") self.assertEqual(cfg.client, "200") self.assertEqual(cfg.user, "MYUSER") self.assertIsNotNone(loaded_from) finally: os.unlink(path) def test_env_overrides_file(self): fd, path = tempfile.mkstemp(suffix=".ini") os.close(fd) try: self._write_config(path, host="file-host", client="100", user="file-user", password="file-pass") env = self._clean_env() env["SAP_HOST"] = "env-host" env["SAP_USER"] = "env-user" env["SAP_PASSWORD"] = "env-pass" with patch.dict(os.environ, env, clear=True): cfg, _ = load_config(config_path=path) self.assertEqual(cfg.host, "env-host") self.assertEqual(cfg.user, "env-user") finally: os.unlink(path) def test_missing_host_raises_config_error(self): fd, path = tempfile.mkstemp(suffix=".ini") os.close(fd) try: self._write_config(path, client="100", user="u", password="p") with patch.dict(os.environ, self._clean_env(), clear=True): with self.assertRaises(ConfigError): load_config(config_path=path) finally: os.unlink(path) def test_profile_parameter(self): fd, path = tempfile.mkstemp(suffix=".ini") os.close(fd) try: self._write_config(path, section="DEV", host="dev-host", client="400", user="DEVUSER", password="devpass") with patch.dict(os.environ, self._clean_env(), clear=True): cfg, _ = load_config(config_path=path, profile="DEV") self.assertEqual(cfg.host, "dev-host") self.assertEqual(cfg.client, "400") finally: os.unlink(path) # ═══════════════════════════════════════════════════════════════════ # manifest.py # ═══════════════════════════════════════════════════════════════════ class TestManifestEntry(unittest.TestCase): """ManifestEntry 序列化 / 反序列化。""" def test_to_dict_defaults(self): entry = ManifestEntry(name="ZFOO", type="report", file="reports/zfoo.abap") d = entry.to_dict() self.assertEqual(d["type"], "report") self.assertEqual(d["file"], "reports/zfoo.abap") self.assertEqual(d["system_status"], "not_exists") self.assertIsNone(d["corr_nr"]) self.assertEqual(d["depends_on"], []) self.assertIsNone(d["last_sync"]) self.assertEqual(d["last_sync_result"], "pending") def test_to_dict_full(self): entry = ManifestEntry( name="ZBAR", type="class", file="classes/zbar.abap", system_status="active", corr_nr="NR001", depends_on=["ZFOO"], last_sync="2026-01-01T00:00:00", last_sync_result="success", ) d = entry.to_dict() self.assertEqual(d["system_status"], "active") self.assertEqual(d["corr_nr"], "NR001") self.assertEqual(d["depends_on"], ["ZFOO"]) self.assertEqual(d["last_sync"], "2026-01-01T00:00:00") self.assertEqual(d["last_sync_result"], "success") def test_from_dict_roundtrip(self): original = ManifestEntry( name="ZRT", type="table", file="tables/zrt.abap", system_status="inactive", corr_nr="C002", depends_on=["ZA", "ZB"], last_sync="2026-06-01T12:00:00", last_sync_result="failed", ) d = original.to_dict() restored = ManifestEntry.from_dict("ZRT", d) self.assertEqual(restored.name, original.name) self.assertEqual(restored.type, original.type) self.assertEqual(restored.file, original.file) self.assertEqual(restored.system_status, original.system_status) self.assertEqual(restored.corr_nr, original.corr_nr) self.assertEqual(restored.depends_on, original.depends_on) self.assertEqual(restored.last_sync, original.last_sync) self.assertEqual(restored.last_sync_result, original.last_sync_result) def test_from_dict_defaults(self): entry = ManifestEntry.from_dict("ZNEW", {"type": "domain", "file": "domains/znew.abap"}) self.assertEqual(entry.name, "ZNEW") self.assertEqual(entry.system_status, "not_exists") self.assertIsNone(entry.corr_nr) self.assertEqual(entry.depends_on, []) self.assertIsNone(entry.last_sync) self.assertEqual(entry.last_sync_result, "pending") class TestManifest(unittest.TestCase): """Manifest load/save 往返和 CRUD。""" def test_save_and_load_roundtrip(self): with tempfile.TemporaryDirectory() as tmp: manifest = Manifest( version=CURRENT_VERSION, last_init="2026-01-01T00:00:00", last_refresh=None, objects={ "ZFOO": ManifestEntry( name="ZFOO", type="report", file="reports/zfoo.abap", system_status="active", last_sync_result="success", ), "ZBAR": ManifestEntry( name="ZBAR", type="class", file="classes/zbar.abap", depends_on=["ZFOO"], ), }, _project_path=tmp, ) manifest.save() filepath = os.path.join(tmp, MANIFEST_FILENAME) self.assertTrue(os.path.isfile(filepath)) loaded = Manifest.load(tmp) self.assertEqual(loaded.version, CURRENT_VERSION) self.assertEqual(len(loaded.objects), 2) self.assertIn("ZFOO", loaded.objects) self.assertIn("ZBAR", loaded.objects) self.assertEqual(loaded.objects["ZFOO"].type, "report") self.assertEqual(loaded.objects["ZBAR"].depends_on, ["ZFOO"]) def test_load_missing_file_raises(self): with tempfile.TemporaryDirectory() as tmp: with self.assertRaises(FileNotFoundError): Manifest.load(tmp) def test_upsert_and_remove(self): m = Manifest() entry = ManifestEntry(name="Z1", type="report", file="reports/z1.abap") m.upsert(entry) self.assertIn("Z1", m.objects) self.assertEqual(m.get("Z1").type, "report") entry2 = ManifestEntry(name="Z1", type="class", file="classes/z1.abap") m.upsert(entry2) self.assertEqual(m.get("Z1").type, "class") self.assertTrue(m.remove("Z1")) self.assertIsNone(m.get("Z1")) self.assertFalse(m.remove("Z1")) def test_pending_objects(self): m = Manifest() m.upsert(ManifestEntry(name="A", type="report", file="r/a.abap", system_status="active", last_sync_result="success")) m.upsert(ManifestEntry(name="B", type="class", file="c/b.abap", system_status="not_exists", last_sync_result="pending")) m.upsert(ManifestEntry(name="C", type="table", file="t/c.abap", system_status="active", last_sync_result="failed")) pending = m.pending_objects() names = [e.name for e in pending] self.assertIn("B", names) self.assertIn("C", names) self.assertNotIn("A", names) def test_active_up_to_date_objects(self): m = Manifest() m.upsert(ManifestEntry(name="A", type="report", file="r/a.abap", system_status="active", last_sync_result="success")) m.upsert(ManifestEntry(name="B", type="class", file="c/b.abap", system_status="not_exists", last_sync_result="pending")) active = m.active_up_to_date_objects() self.assertEqual(len(active), 1) self.assertEqual(active[0].name, "A") def test_file_path(self): m = Manifest(_project_path="/project") entry = ManifestEntry(name="Z1", type="report", file="reports/z1.abap") fp = m.file_path(entry) expected = os.path.join("/project", "reports", "z1.abap") self.assertEqual(fp, expected) class TestInitManifest(unittest.TestCase): """init_manifest 创建空清单。""" def test_creates_manifest_with_init_time(self): m = init_manifest("/tmp/fake") self.assertEqual(m.version, CURRENT_VERSION) self.assertIsNotNone(m.last_init) self.assertEqual(len(m.objects), 0) self.assertEqual(m._project_path, "/tmp/fake") # ═══════════════════════════════════════════════════════════════════ # scanner.py # ═══════════════════════════════════════════════════════════════════ class TestScanProject(unittest.TestCase): """scan_project 目录扫描。""" def _make_project(self, tmp: str) -> None: """创建模拟项目目录结构。""" # reports/zrpt001.abap os.makedirs(os.path.join(tmp, "reports")) with open(os.path.join(tmp, "reports", "zrpt001.abap"), "w") as f: f.write("REPORT zrpt001.\n") # reports/.hidden.abap — 应被跳过 with open(os.path.join(tmp, "reports", ".hidden.abap"), "w") as f: f.write("REPORT hidden.\n") # reports/notes.txt — 非 .abap 应被跳过 with open(os.path.join(tmp, "reports", "notes.txt"), "w") as f: f.write("notes\n") # classes/zcl_demo.abap os.makedirs(os.path.join(tmp, "classes")) with open(os.path.join(tmp, "classes", "zcl_demo.abap"), "w") as f: f.write("CLASS zcl_demo DEFINITION.\n") # functions/zgroup/zfunc1.abap os.makedirs(os.path.join(tmp, "functions", "zgroup")) with open(os.path.join(tmp, "functions", "zgroup", "zfunc1.abap"), "w") as f: f.write("FUNCTION zfunc1.\n") # functions/zgroup/zfunc2.abap with open(os.path.join(tmp, "functions", "zgroup", "zfunc2.abap"), "w") as f: f.write("FUNCTION zfunc2.\n") # functions/.hidden_grp/func.abap — 隐藏目录应被跳过 os.makedirs(os.path.join(tmp, "functions", ".hidden_grp")) with open(os.path.join(tmp, "functions", ".hidden_grp", "func.abap"), "w") as f: f.write("FUNCTION func.\n") # domains/zdom_test.abap os.makedirs(os.path.join(tmp, "domains")) with open(os.path.join(tmp, "domains", "zdom_test.abap"), "w") as f: f.write("DOMAIN zdom_test.\n") def test_scan_finds_all_objects(self): with tempfile.TemporaryDirectory() as tmp: self._make_project(tmp) results = scan_project(tmp) names = {r.name for r in results} self.assertIn("ZRPT001", names) self.assertNotIn("HIDDEN", names) self.assertIn("ZCL_DEMO", names) self.assertIn("ZGROUP/ZFUNC1", names) self.assertIn("ZGROUP/ZFUNC2", names) self.assertNotIn("HIDDEN_GRP/FUNC", names) self.assertIn("ZDOM_TEST", names) def test_scan_types(self): with tempfile.TemporaryDirectory() as tmp: self._make_project(tmp) results = scan_project(tmp) type_map = {r.name: r.type for r in results} self.assertEqual(type_map.get("ZRPT001"), "report") self.assertEqual(type_map.get("ZCL_DEMO"), "class") self.assertEqual(type_map.get("ZGROUP/ZFUNC1"), "function") self.assertEqual(type_map.get("ZGROUP/ZFUNC2"), "function") self.assertEqual(type_map.get("ZDOM_TEST"), "domain") def test_scan_file_paths(self): with tempfile.TemporaryDirectory() as tmp: self._make_project(tmp) results = scan_project(tmp) file_map = {r.name: r.file for r in results} self.assertEqual(file_map["ZRPT001"], "reports/zrpt001.abap") self.assertEqual(file_map["ZCL_DEMO"], "classes/zcl_demo.abap") self.assertEqual(file_map["ZGROUP/ZFUNC1"], "functions/zgroup/zfunc1.abap") self.assertEqual(file_map["ZDOM_TEST"], "domains/zdom_test.abap") def _make_adt_project(self, tmp: str) -> None: """创建 ADT/abapGit 风格项目（`..abap`，项目仓命名规范形态）。""" os.makedirs(os.path.join(tmp, "src", "ZINT.package", "class")) cls_dir = os.path.join(tmp, "src", "ZINT.package", "class") # ADT 式：zcl_demo.clas.abap with open(os.path.join(cls_dir, "zcl_demo.clas.abap"), "w") as f: f.write("CLASS zcl_demo DEFINITION.\n") # 无后缀式（历史写法，仍须兼容）：zcl_plain.abap with open(os.path.join(cls_dir, "zcl_plain.abap"), "w") as f: f.write("CLASS zcl_plain DEFINITION.\n") # 后缀与目录类型不匹配：不得被剥（否则对象名会被错改） with open(os.path.join(cls_dir, "zcl_weird.prog.abap"), "w") as f: f.write("CLASS zcl_weird DEFINITION.\n") def test_scan_adt_style_filenames(self): """ADT 风格 `.clas.abap` 必须解析为对象名 ZCL_DEMO（不带 .CLAS）。""" with tempfile.TemporaryDirectory() as tmp: self._make_adt_project(tmp) results = scan_project(tmp) names = {r.name for r in results} self.assertIn("ZCL_DEMO", names) self.assertNotIn("ZCL_DEMO.CLAS", names) # 无后缀式兼容 self.assertIn("ZCL_PLAIN", names) # 目录类型不匹配的后缀不剥 self.assertIn("ZCL_WEIRD.PROG", names) def test_scan_adt_style_file_paths(self): """ADT 风格项目下 file 字段仍为仓内相对路径。""" with tempfile.TemporaryDirectory() as tmp: self._make_adt_project(tmp) file_map = {r.name: r.file for r in scan_project(tmp)} self.assertEqual( file_map["ZCL_DEMO"], "src/ZINT.package/class/zcl_demo.clas.abap", ) def test_strip_adt_suffix_unit(self): """后缀剥离函数单测：只剥匹配目录类型的那一个后缀。""" from sapcli.scanner import _strip_adt_suffix self.assertEqual(_strip_adt_suffix("zcl_demo.clas", "class"), "zcl_demo") self.assertEqual(_strip_adt_suffix("zcl_demo", "class"), "zcl_demo") self.assertEqual(_strip_adt_suffix("zcl_demo.prog", "class"), "zcl_demo.prog") self.assertEqual(_strip_adt_suffix("zrpt1.prog", "report"), "zrpt1") self.assertEqual(_strip_adt_suffix("zdom.doma", "domain"), "zdom") # 后缀即全部文件名时不得剥成空串 self.assertEqual(_strip_adt_suffix("clas", "class"), "clas") def test_scan_empty_directory(self): with tempfile.TemporaryDirectory() as tmp: results = scan_project(tmp) self.assertEqual(results, []) def test_scanned_object_dataclass(self): obj = ScannedObject(name="ZTEST", type="report", file="reports/ztest.abap") self.assertEqual(obj.name, "ZTEST") self.assertEqual(obj.type, "report") self.assertEqual(obj.file, "reports/ztest.abap") # ═══════════════════════════════════════════════════════════════════ # sorter.py # ═══════════════════════════════════════════════════════════════════ class TestTopologicalSort(unittest.TestCase): """topological_sort 按类型优先级和依赖排序。""" def _entry(self, name: str, obj_type: str, depends_on: list[str] | None = None) -> ManifestEntry: return ManifestEntry( name=name, type=obj_type, file=f"{obj_type}s/{name.lower()}.abap", depends_on=depends_on or [], ) def test_empty_list(self): self.assertEqual(topological_sort([]), []) def test_sorts_by_type_priority(self): """无依赖时，按 TYPE_PRIORITY 数值从小到大排序。""" objs = [ self._entry("ZCL1", "class"), self._entry("ZDOM1", "domain"), self._entry("ZRPT1", "report"), self._entry("ZTAB1", "table"), ] result = topological_sort(objs) names = [o.name for o in result] self.assertEqual(names, ["ZDOM1", "ZTAB1", "ZCL1", "ZRPT1"]) def test_respects_depends_on(self): """有依赖时，被依赖的对象排在前面。""" objs = [ self._entry("ZRPT1", "report", depends_on=["ZCL1"]), self._entry("ZCL1", "class", depends_on=["ZTAB1"]), self._entry("ZTAB1", "table"), ] result = topological_sort(objs) names = [o.name for o in result] idx_tab = names.index("ZTAB1") idx_cl = names.index("ZCL1") idx_rpt = names.index("ZRPT1") self.assertLess(idx_tab, idx_cl) self.assertLess(idx_cl, idx_rpt) def test_cyclic_dependency_raises(self): """循环依赖应抛出 CyclicDependencyError。""" objs = [ self._entry("A", "report", depends_on=["B"]), self._entry("B", "class", depends_on=["A"]), ] with self.assertRaises(CyclicDependencyError): topological_sort(objs) def test_external_dependency_ignored(self): """depends_on 引用不在列表中的对象时，忽略该依赖。""" objs = [ self._entry("ZRPT1", "report", depends_on=["EXTERNAL"]), ] result = topological_sort(objs) self.assertEqual(len(result), 1) self.assertEqual(result[0].name, "ZRPT1") def test_unknown_type_gets_default_priority(self): """未注册类型应获得默认优先级 (90)。""" objs = [ self._entry("ZUNKNOWN", "weird_type"), self._entry("ZRPT1", "report"), ] result = topological_sort(objs) names = [o.name for o in result] self.assertEqual(names, ["ZRPT1", "ZUNKNOWN"]) def test_diamond_dependency(self): """菱形依赖: A → B, A → C, B → D, C → D。D 最先。""" objs = [ self._entry("A", "report", depends_on=["B", "C"]), self._entry("B", "class", depends_on=["D"]), self._entry("C", "function", depends_on=["D"]), self._entry("D", "domain"), ] result = topological_sort(objs) names = [o.name for o in result] idx_d = names.index("D") idx_b = names.index("B") idx_c = names.index("C") idx_a = names.index("A") self.assertLess(idx_d, idx_b) self.assertLess(idx_d, idx_c) self.assertLess(idx_b, idx_a) self.assertLess(idx_c, idx_a) # ═══════════════════════════════════════════════════════════════════ # auth.py # ═══════════════════════════════════════════════════════════════════ class TestServiceName(unittest.TestCase): """_service_name 格式。""" def test_format(self): from sapcli.auth import _service_name self.assertEqual(_service_name("myhost.com", "100"), "sap-cli:myhost.com:100") def test_format_empty_client(self): from sapcli.auth import _service_name self.assertEqual(_service_name("host", ""), "sap-cli:host:") class _AuthTestBase(unittest.TestCase): """auth 测试基类，mock keyring 模块。""" def _patch_keyring(self): """将 sapcli.auth / sapcli.password 的 _keyring / _KEYRING_AVAILABLE 替换为 mock。""" import sapcli.auth as auth_mod import sapcli.password as pwd_mod self.mock_kr = MagicMock() # mock keyring module with errors attribute self.mock_kr_mod = MagicMock() self.mock_kr_mod.errors.PasswordDeleteError = type("PasswordDeleteError", (Exception,), {}) # 保存原始值以便恢复 self._orig_keyring = getattr(auth_mod, "_keyring", None) self._orig_available = getattr(auth_mod, "_KEYRING_AVAILABLE", None) self._orig_kr_mod = getattr(auth_mod, "_keyring_mod", None) self._had_kr_mod = hasattr(auth_mod, "_keyring_mod") self._orig_pwd_keyring = getattr(pwd_mod, "_keyring", None) self._orig_pwd_available = getattr(pwd_mod, "_KEYRING_AVAILABLE", None) self._orig_pwd_kr_mod = getattr(pwd_mod, "_keyring_mod", None) self._had_pwd_kr_mod = hasattr(pwd_mod, "_keyring_mod") # auth 模块 auth_mod._keyring = self.mock_kr auth_mod._KEYRING_AVAILABLE = True auth_mod._keyring_mod = self.mock_kr_mod # password 模块（resolve_password / get_password 实际所在） pwd_mod._keyring = self.mock_kr pwd_mod._KEYRING_AVAILABLE = True pwd_mod._keyring_mod = self.mock_kr_mod self.addCleanup(self._restore_keyring) def _restore_keyring(self): import sapcli.auth as auth_mod import sapcli.password as pwd_mod auth_mod._keyring = self._orig_keyring auth_mod._KEYRING_AVAILABLE = self._orig_available if self._had_kr_mod: auth_mod._keyring_mod = self._orig_kr_mod elif hasattr(auth_mod, "_keyring_mod"): del auth_mod._keyring_mod pwd_mod._keyring = self._orig_pwd_keyring pwd_mod._KEYRING_AVAILABLE = self._orig_pwd_available if self._had_pwd_kr_mod: pwd_mod._keyring_mod = self._orig_pwd_kr_mod elif hasattr(pwd_mod, "_keyring_mod"): del pwd_mod._keyring_mod class TestGetPassword(_AuthTestBase): """get_password 从 keyring 读取密码。""" def setUp(self): self._patch_keyring() def test_returns_password(self): from sapcli.auth import get_password self.mock_kr.get_password.return_value = "secret123" result = get_password("host", "100", "user1") self.assertEqual(result, "secret123") self.mock_kr.get_password.assert_called_once_with("sap-cli:host:100", "user1") def test_returns_none_on_exception(self): from sapcli.auth import get_password self.mock_kr.get_password.side_effect = Exception("boom") result = get_password("host", "100", "user1") self.assertIsNone(result) def test_returns_none_when_unavailable(self): from sapcli.auth import get_password with patch("sapcli.password._KEYRING_AVAILABLE", False): result = get_password("host", "100", "user1") self.assertIsNone(result) class TestSetPassword(_AuthTestBase): """set_password 存储密码到 keyring。""" def setUp(self): self._patch_keyring() def test_success(self): from sapcli.auth import set_password result = set_password("host", "100", "user1", "pass") self.assertTrue(result) self.mock_kr.set_password.assert_called_once_with("sap-cli:host:100", "user1", "pass") def test_failure_returns_false(self): from sapcli.auth import set_password self.mock_kr.set_password.side_effect = Exception("fail") result = set_password("host", "100", "user1", "pass") self.assertFalse(result) def test_unavailable_returns_false(self): from sapcli.auth import set_password with patch("sapcli.auth._KEYRING_AVAILABLE", False): result = set_password("host", "100", "user1", "pass") self.assertFalse(result) class TestDeletePassword(_AuthTestBase): """delete_password 从 keyring 删除密码。""" def setUp(self): self._patch_keyring() def test_success(self): from sapcli.auth import delete_password result = delete_password("host", "100", "user1") self.assertTrue(result) self.mock_kr.delete_password.assert_called_once_with("sap-cli:host:100", "user1") def test_failure_returns_false(self): from sapcli.auth import delete_password self.mock_kr.delete_password.side_effect = Exception("fail") result = delete_password("host", "100", "user1") self.assertFalse(result) def test_unavailable_returns_false(self): from sapcli.auth import delete_password with patch("sapcli.auth._KEYRING_AVAILABLE", False): result = delete_password("host", "100", "user1") self.assertFalse(result) class TestResolvePassword(_AuthTestBase): """resolve_password 按优先级解析密码。""" def setUp(self): self._patch_keyring() def test_env_password_takes_priority(self): from sapcli.auth import resolve_password self.mock_kr.get_password.return_value = "kr_pass" result = resolve_password("h", "100", "u", config_password="cfg_pass", env_password="env_pass") self.assertEqual(result, "env_pass") def test_keyring_over_config(self): from sapcli.auth import resolve_password self.mock_kr.get_password.return_value = "kr_pass" result = resolve_password("h", "100", "u", config_password="cfg_pass", env_password="") self.assertEqual(result, "kr_pass") def test_config_as_fallback(self): from sapcli.auth import resolve_password self.mock_kr.get_password.return_value = None result = resolve_password("h", "100", "u", config_password="cfg_pass", env_password="") self.assertEqual(result, "cfg_pass") def test_all_empty(self): from sapcli.auth import resolve_password self.mock_kr.get_password.return_value = None result = resolve_password("h", "100", "u", config_password="", env_password="") self.assertEqual(result, "") class TestCmdAuthStatus(unittest.TestCase): """cmd_auth_status 输出。""" def test_prints_keyring_available(self): from sapcli.auth import cmd_auth_status mock_backend = MagicMock() mock_backend.__class__ = type("TestBackend", (), {}) with patch("sapcli.auth._KEYRING_AVAILABLE", True), \ patch("sapcli.auth._keyring") as mock_kr: mock_kr.get_keyring.return_value = mock_backend with patch("sys.stdout", new_callable=io.StringIO) as out: cmd_auth_status(argparse.Namespace()) output = out.getvalue() self.assertIn("keyring", output) self.assertIn("可用", output) def test_prints_keyring_unavailable(self): from sapcli.auth import cmd_auth_status with patch("sapcli.auth._KEYRING_AVAILABLE", False): with patch("sys.stdout", new_callable=io.StringIO) as out: cmd_auth_status(argparse.Namespace()) output = out.getvalue() self.assertIn("keyring", output) self.assertIn("不可用", output) class TestCmdAuthLogin(unittest.TestCase): """cmd_auth_login 交互式登录。""" def test_login_stores_password(self): from sapcli.auth import cmd_auth_login fake_cfg = SAPConfig(host="h", client="100", user="u", password="p") args = argparse.Namespace(config=None) with patch("sapcli.config.load_config", return_value=(fake_cfg, "/fake")), \ patch("sapcli.auth.getpass") as mock_gp_mod, \ patch("sapcli.auth.set_password", return_value=True) as mock_sp, \ patch("sys.stdout", new_callable=io.StringIO): mock_gp_mod.getpass.return_value = "mypassword" cmd_auth_login(args) mock_sp.assert_called_once_with("h", "100", "u", "mypassword") def test_login_empty_password_aborts(self): from sapcli.auth import cmd_auth_login fake_cfg = SAPConfig(host="h", client="100", user="u", password="p") with patch("sapcli.config.load_config", return_value=(fake_cfg, "/fake")), \ patch("sapcli.auth.getpass.getpass", return_value=""), \ patch("sapcli.auth.set_password") as mock_sp, \ patch("sys.stdout", new_callable=io.StringIO) as out: cmd_auth_login(argparse.Namespace(config=None)) # 不应调用 set_password mock_sp.assert_not_called() self.assertIn("不能为空", out.getvalue()) class TestCmdAuthLogout(unittest.TestCase): """cmd_auth_logout 删除密码。""" def test_logout_deletes_password(self): from sapcli.auth import cmd_auth_logout fake_cfg = SAPConfig(host="h", client="100", user="u", password="p") with patch("sapcli.config.load_config", return_value=(fake_cfg, "/fake")), \ patch("sapcli.auth.delete_password", return_value=True) as mock_dp, \ patch("sys.stdout", new_callable=io.StringIO) as out: cmd_auth_logout(argparse.Namespace(config=None)) mock_dp.assert_called_once_with("h", "100", "u") self.assertIn("删除", out.getvalue()) class TestScanProjectThreeLayer(unittest.TestCase): """scan_project 三层布局（src/<开发包>/<对象类型>/）+ 单复数目录名。""" def _make_project(self, tmp: str) -> None: """src/TMP/<类型>/ 三层布局。""" for sub in ( "src/TMP/report", "src/TMP/class", "src/TMP/interface", "src/TMP/domain", "src/TMP/structure", "src/TMP/tabletype", ): os.makedirs(os.path.join(tmp, *sub.split("/"))) def put(rel: str, body: str) -> None: with open(os.path.join(tmp, *rel.split("/")), "w") as f: f.write(body) put("src/TMP/report/zrpt001.abap", "REPORT zrpt001.\n") put("src/TMP/report/notes.txt", "skip me\n") # 非 .abap 跳过 put("src/TMP/class/zcl_demo.abap", "CLASS zcl_demo DEFINITION.\n") put("src/TMP/interface/zif_demo.abap", "INTERFACE zif_demo.\n") put("src/TMP/domain/zdom_demo.abap", "DOMAIN zdom_demo.\n") put("src/TMP/structure/zst_demo.abap", "STRUCTURE zst_demo.\n") put("src/TMP/tabletype/ztt_demo.abap", "TABLETYPE ztt_demo.\n") def _make_funcs(self, tmp: str) -> None: """函数模块两种命名。""" os.makedirs(os.path.join(tmp, "src", "TMP", "function", "zfg_one")) with open(os.path.join(tmp, "src", "TMP", "function", "zfg_one", "zfm_a.abap"), "w") as f: f.write("FUNCTION zfm_a.\n") with open(os.path.join(tmp, "src", "TMP", "function", "zfg_two.fugr.zfm_b.abap"), "w") as f: f.write("FUNCTION zfm_b.\n") def test_three_layer_finds_objects(self): with tempfile.TemporaryDirectory() as tmp: self._make_project(tmp) results = scan_project(tmp) by_name = {r.name: r for r in results} self.assertIn("ZRPT001", by_name) self.assertNotIn("NOTES", by_name) self.assertIn("ZCL_DEMO", by_name) self.assertIn("ZIF_DEMO", by_name) self.assertIn("ZDOM_DEMO", by_name) self.assertIn("ZST_DEMO", by_name) self.assertIn("ZTT_DEMO", by_name) # 类型识别正确（按目录名，而非文件名） self.assertEqual(by_name["ZCL_DEMO"].type, "class") self.assertEqual(by_name["ZST_DEMO"].type, "structure") self.assertEqual(by_name["ZTT_DEMO"].type, "tabletype") # 相对项目根路径保留三层 self.assertEqual(by_name["ZCL_DEMO"].file, "src/TMP/class/zcl_demo.abap") def test_function_two_naming_styles(self): with tempfile.TemporaryDirectory() as tmp: self._make_funcs(tmp) names = {r.name for r in scan_project(tmp)} self.assertIn("ZFG_ONE/ZFM_A", names) # 子目录式 self.assertIn("ZFG_TWO/ZFM_B", names) # ADT 式 .fugr. def test_plural_and_case_insensitive(self): """classes/ 与 class/ 等价，目录名大小写不敏感。""" with tempfile.TemporaryDirectory() as tmp: os.makedirs(os.path.join(tmp, "src", "TMP", "CLASSES")) with open(os.path.join(tmp, "src", "TMP", "CLASSES", "zcl_x.abap"), "w") as f: f.write("CLASS zcl_x DEFINITION.\n") res = scan_project(tmp) self.assertEqual([(r.name, r.type) for r in res], [("ZCL_X", "class")]) def test_backward_compatible_flat_layout(self): """旧的扁平布局仍可扫描。""" with tempfile.TemporaryDirectory() as tmp: os.makedirs(os.path.join(tmp, "reports")) with open(os.path.join(tmp, "reports", "zold.abap"), "w") as f: f.write("REPORT zold.\n") os.makedirs(os.path.join(tmp, "classes")) with open(os.path.join(tmp, "classes", "zcl_old.abap"), "w") as f: f.write("CLASS zcl_old DEFINITION.\n") names = {r.name for r in scan_project(tmp)} self.assertIn("ZOLD", names) self.assertIn("ZCL_OLD", names) def test_no_duplicate_when_both_layouts(self): """同一对象在两处出现只保留一次。""" with tempfile.TemporaryDirectory() as tmp: os.makedirs(os.path.join(tmp, "src", "TMP", "report")) os.makedirs(os.path.join(tmp, "reports")) for d in ("src/TMP/report", "reports"): with open(os.path.join(tmp, *d.split("/"), "zdup.abap"), "w") as f: f.write("REPORT zdup.\n") res = scan_project(tmp) self.assertEqual(len([r for r in res if r.name == "ZDUP"]), 1) if __name__ == "__main__": unittest.main()