""" sap-cli 单元测试(v3 — 匹配重构后 API) 运行: python tests/test_sapcli.py """ import io import json import os import sys import tempfile import unittest from unittest.mock import MagicMock, patch # 将项目根目录加入 sys.path,支持从任意位置运行测试 sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "assets")) # ═══════════════════════════════════════════ # TestTypes # ═══════════════════════════════════════════ class TestTypes(unittest.TestCase): """对象类型注册与名称解析。""" def test_type_registry(self): from sapcli.types import all_type_keys, get_type_config keys = all_type_keys() self.assertGreaterEqual(len(keys), 10) for k in keys: cfg = get_type_config(k) self.assertIsNotNone(cfg) self.assertTrue(cfg.obj_uri_template) def test_parse_report_name(self): from sapcli.types import parse_object_name parsed = parse_object_name("ZMY_REPORT", "report") self.assertEqual(parsed.display_name, "ZMY_REPORT") def test_parse_class_name(self): from sapcli.types import parse_object_name parsed = parse_object_name("ZCL_MY_CLASS", "class") self.assertEqual(parsed.display_name, "ZCL_MY_CLASS") def test_parse_function_name(self): from sapcli.types import parse_object_name parsed = parse_object_name("ZGROUP/Z_MY_FUNC", "function") self.assertIn("z_my_func", parsed.obj_uri) def test_parse_function_name_no_slash(self): from sapcli.types import parse_object_name from sapcli.exceptions import InvalidNameError with self.assertRaises(InvalidNameError): parse_object_name("Z_MY_FUNC", "function") def test_new_types_registered(self): from sapcli.types import all_type_keys, get_type_config keys = all_type_keys() # Check that all 10 core types are registered expected = ["report", "class", "function", "functiongroup", "interface", "domain", "dataelement", "table", "structure", "tabletype"] for t in expected: self.assertIn(t, keys) cfg = get_type_config(t) self.assertIsNotNone(cfg) def test_r3tr_object_code(self): from sapcli.types import get_r3tr_object_code # 有独立 R3TR 代码 self.assertEqual(get_r3tr_object_code("class"), "CLAS") self.assertEqual(get_r3tr_object_code("report"), "PROG") self.assertEqual(get_r3tr_object_code("table"), "TABL") self.assertEqual(get_r3tr_object_code("domain"), "DOMA") self.assertEqual(get_r3tr_object_code("cdsview"), "DDLS") # 非独立 R3TR 对象(属父对象) self.assertIsNone(get_r3tr_object_code("function")) self.assertIsNone(get_r3tr_object_code("include")) # 未知类型 self.assertIsNone(get_r3tr_object_code("not_a_type")) # ═══════════════════════════════════════════ # TestExceptions # ═══════════════════════════════════════════ class TestExceptions(unittest.TestCase): def test_exceptions(self): from sapcli.exceptions import ( SapCliError, ConfigError, LoginError, LockError, ActivationError, CreateError, DeleteError, SyntaxCheckError, InvalidNameError, ObjectNotFoundError, CyclicDependencyError, ) for ExcCls in [SapCliError, ConfigError, LoginError, LockError, ActivationError, CreateError, DeleteError, SyntaxCheckError, InvalidNameError]: e = ExcCls("test error") self.assertIsInstance(e, SapCliError) def test_cyclic_dependency(self): from sapcli.exceptions import CyclicDependencyError e = CyclicDependencyError(["A", "B", "C"]) self.assertIn("A", str(e)) self.assertEqual(e.cycle_members, ["A", "B", "C"]) # ═══════════════════════════════════════════ # TestConfig # ═══════════════════════════════════════════ class TestConfig(unittest.TestCase): def setUp(self): self._saved_env = {k: os.environ.pop(k, None) for k in ["SAP_HOST", "SAP_CLIENT", "SAP_USER", "SAP_PASSWORD", "SAPCLI_PROFILE"]} def tearDown(self): for k, v in self._saved_env.items(): if v is not None: os.environ[k] = v def test_config_from_env(self): from sapcli.config import load_config with patch.dict(os.environ, { "SAP_HOST": "http://test:8000", "SAP_CLIENT": "200", "SAP_USER": "testuser", "SAP_PASSWORD": "testpass", }): cfg, _ = load_config("/nonexistent/config.ini") self.assertEqual(cfg.host, "http://test:8000") self.assertEqual(cfg.client, "200") def test_config_no_file_returns_empty(self): """无配置文件且无环境变量时,load_config 抛出 ConfigError。""" from sapcli.config import load_config from sapcli.exceptions import ConfigError # clear env vars and prevent reading any config file with patch.dict(os.environ, {}, clear=True), \ patch("sapcli.config.os.path.isfile", return_value=False): with self.assertRaises(ConfigError): load_config("/nonexistent/config.ini") def test_multi_profile(self): from sapcli.config import load_config with tempfile.NamedTemporaryFile(mode="w", suffix=".ini", delete=False, encoding="utf-8") as f: f.write("[SAP]\nhost=http://dev:8000\nclient=100\nuser=devuser\npassword=devpass\n") f.write("[QAS]\nhost=http://qas:8000\nclient=200\nuser=qasuser\npassword=qaspass\n") f.write("[PRD]\nhost=http://prd:8000\nclient=300\nuser=prduser\npassword=prdpass\n") tmppath = f.name try: # 默认 profile=SAP cfg, _ = load_config(tmppath) self.assertEqual(cfg.host, "http://dev:8000") # 指定 QAS cfg, _ = load_config(tmppath, profile="QAS") self.assertEqual(cfg.host, "http://qas:8000") self.assertEqual(cfg.client, "200") # 指定 PRD cfg, _ = load_config(tmppath, profile="PRD") self.assertEqual(cfg.host, "http://prd:8000") finally: os.unlink(tmppath) def test_profile_fallback(self): from sapcli.config import load_config from sapcli.exceptions import ConfigError with tempfile.NamedTemporaryFile(mode="w", suffix=".ini", delete=False, encoding="utf-8") as f: f.write("[SAP]\nhost=http://dev:8000\nclient=100\nuser=devuser\npassword=devpass\n") tmppath = f.name try: # 不存在的 profile → section 为空 → ConfigError with self.assertRaises(ConfigError): load_config(tmppath, profile="NONEXISTENT") finally: os.unlink(tmppath) # ═══════════════════════════════════════════ # TestManifest # ═══════════════════════════════════════════ class TestManifest(unittest.TestCase): def _make_manifest(self, td): from sapcli.manifest import Manifest m = Manifest(_project_path=td) return m def _make_entry(self, name, obj_type, file, **kw): from sapcli.manifest import ManifestEntry return ManifestEntry(name=name, type=obj_type, file=file, **kw) def test_manifest_add_and_get(self): with tempfile.TemporaryDirectory() as td: m = self._make_manifest(td) m.upsert(self._make_entry("Z_TEST", "report", "reports/z_test.abap")) entry = m.get("Z_TEST") self.assertIsNotNone(entry) self.assertEqual(entry.type, "report") def test_manifest_remove(self): with tempfile.TemporaryDirectory() as td: m = self._make_manifest(td) m.upsert(self._make_entry("Z_A", "domain", "domains/z_a.abap")) m.remove("Z_A") self.assertIsNone(m.get("Z_A")) def test_manifest_pending(self): with tempfile.TemporaryDirectory() as td: m = self._make_manifest(td) m.upsert(self._make_entry("Z_A", "domain", "domains/z_a.abap", system_status="active", last_sync_result="success")) m.upsert(self._make_entry("Z_B", "report", "reports/z_b.abap", system_status="inactive")) m.upsert(self._make_entry("Z_C", "class", "classes/z_c.abap", system_status="not_exists")) pending = m.pending_objects() names = [p.name for p in pending] self.assertIn("Z_B", names) self.assertIn("Z_C", names) self.assertNotIn("Z_A", names) def test_manifest_persistence(self): with tempfile.TemporaryDirectory() as td: m = self._make_manifest(td) m.upsert(self._make_entry("Z_TEST", "report", "reports/z_test.abap")) m.save() self.assertTrue(os.path.isfile(os.path.join(td, "manifest.json"))) def test_manifest_load_empty(self): with tempfile.TemporaryDirectory() as td: m = self._make_manifest(td) self.assertEqual(len(m.objects), 0) # ═══════════════════════════════════════════ # TestScanner # ═══════════════════════════════════════════ class TestScanner(unittest.TestCase): def test_scanner_basic(self): from sapcli.scanner import scan_project with tempfile.TemporaryDirectory() as td: os.makedirs(os.path.join(td, "reports")) with open(os.path.join(td, "reports", "zmy_report.abap"), "w") as f: f.write("WRITE: / 'Hello'.") result = scan_project(td) self.assertEqual(len(result), 1) self.assertEqual(result[0].name, "ZMY_REPORT") self.assertEqual(result[0].type, "report") def test_scanner_empty_dir(self): from sapcli.scanner import scan_project with tempfile.TemporaryDirectory() as td: result = scan_project(td) self.assertEqual(len(result), 0) # ═══════════════════════════════════════════ # TestSorter # ═══════════════════════════════════════════ class TestSorter(unittest.TestCase): def _make_entry(self, name, obj_type, deps=None): from sapcli.manifest import ManifestEntry return ManifestEntry(name=name, type=obj_type, file=f"{name}.abap", depends_on=deps or []) def test_sorter_basic(self): from sapcli.sorter import topological_sort objects = [ self._make_entry("Z_B", "class", ["Z_A"]), self._make_entry("Z_A", "domain", []), ] result = topological_sort(objects) names = [o.name for o in result] self.assertLess(names.index("Z_A"), names.index("Z_B")) def test_sorter_no_deps(self): from sapcli.sorter import topological_sort objects = [ self._make_entry("Z_A", "domain"), self._make_entry("Z_B", "report"), ] result = topological_sort(objects) self.assertEqual(len(result), 2) def test_sorter_cycle_detection(self): from sapcli.sorter import topological_sort from sapcli.exceptions import CyclicDependencyError objects = [ self._make_entry("Z_A", "class", ["Z_B"]), self._make_entry("Z_B", "class", ["Z_A"]), ] with self.assertRaises(CyclicDependencyError): topological_sort(objects) def test_sorter_external_dep_ignored(self): from sapcli.sorter import topological_sort objects = [self._make_entry("Z_A", "domain", ["Z_EXTERNAL"])] result = topological_sort(objects) self.assertEqual(len(result), 1) def test_sorter_empty(self): from sapcli.sorter import topological_sort self.assertEqual(topological_sort([]), []) # ═══════════════════════════════════════════ # TestDdic # ═══════════════════════════════════════════ class TestDdic(unittest.TestCase): def test_ddic_domain_xml(self): from sapcli.ddic import DomainDefinition d = DomainDefinition(datatype="CHAR", length=10) xml = d.to_xml("Z_TEST", "Test domain") self.assertIn("Z_TEST", xml) self.assertIn("Test domain", xml) self.assertIn("CHAR", xml) def test_ddic_domain_fix_values(self): from sapcli.ddic import DomainDefinition d = DomainDefinition( datatype="CHAR", length=1, fix_values=[{"low": "A", "text": "Active"}, {"low": "I", "text": "Inactive"}], ) xml = d.to_xml("Z_STATUS", "Status") self.assertIn("Active", xml) def test_ddic_dataelement_with_domain(self): from sapcli.ddic import DataElementDefinition de = DataElementDefinition(domain_name="Z_STATUS") xml = de.to_xml("Z_STATUS_DE", "Status element", "$TMP") self.assertIn("Z_STATUS", xml) def test_ddic_dataelement_builtin(self): from sapcli.ddic import DataElementDefinition de = DataElementDefinition(datatype="CHAR", length=20) xml = de.to_xml("Z_NAME_DE", "Name element", "$TMP") self.assertIn("CHAR", xml) def test_ddic_table_ddl(self): from sapcli.ddic import TableDefinition, TableField t = TableDefinition( fields=[ TableField(name="MANDT", type_name="MANDT"), TableField(name="ID", type_name="Z_ID"), ], ) ddl = t.to_ddl("ZMY_TABLE", "My table") self.assertIn("zmy_table", ddl) self.assertIn("define table", ddl) self.assertIn("MANDT", ddl) def test_ddic_structure_ddl(self): from sapcli.ddic import StructureDefinition, TableField s = StructureDefinition( fields=[ TableField(name="ID", type_name="Z_ID"), TableField(name="NAME", type_name="Z_NAME"), ], ) ddl = s.to_ddl("ZMY_STRUCT", "My struct") self.assertIn("define structure", ddl) def test_ddic_tabletype_xml(self): from sapcli.ddic import TableTypeDefinition tt = TableTypeDefinition(line_type="ZMY_STRUCT") xml = tt.to_xml("ZTY_TABLE", "Table type", "$TMP") self.assertIn("ZMY_STRUCT", xml) self.assertIn("TTYP", xml) def test_ddic_json_builder(self): from sapcli.ddic import build_definition result = build_definition({"name": "Z_TEST", "datatype": "NUMC", "length": 6}, "domain") self.assertIn("NUMC", result) # ═══════════════════════════════════════════ # TestXmlUtils # ═══════════════════════════════════════════ class TestXmlUtils(unittest.TestCase): def test_xml_escape_basic(self): from sapcli.utils.xml_utils import xml_escape self.assertEqual(xml_escape("hello"), "hello") def test_xml_escape_special_chars(self): from sapcli.utils.xml_utils import xml_escape result = xml_escape('') self.assertIn("<script>", result) self.assertIn(""xss"", result) def test_xml_escape_quotes(self): from sapcli.utils.xml_utils import xml_escape result = xml_escape('say "hello" & \'bye\'') self.assertIn(""", result) self.assertIn("&", result) def test_ddl_escape(self): from sapcli.utils.xml_utils import ddl_escape self.assertEqual(ddl_escape("it's O'Brien's"), "it''s O''Brien''s") # ═══════════════════════════════════════════ # TestAuth # ═══════════════════════════════════════════ class TestAuth(unittest.TestCase): def test_resolve_password_env_priority(self): from sapcli.auth import resolve_password with patch("sapcli.password._KEYRING_AVAILABLE", True): result = resolve_password("h", "c", "u", config_password="cfg", env_password="env") self.assertEqual(result, "env") def test_resolve_password_keyring(self): from sapcli.auth import resolve_password with patch("sapcli.password._KEYRING_AVAILABLE", True), \ patch("sapcli.password.get_password", return_value="kr_pass"): result = resolve_password("h", "c", "u", config_password="cfg") self.assertEqual(result, "kr_pass") def test_resolve_password_config_fallback(self): from sapcli.auth import resolve_password with patch("sapcli.password._KEYRING_AVAILABLE", False): result = resolve_password("h", "c", "u", config_password="cfg_pass") self.assertEqual(result, "cfg_pass") def test_keyring_unavailable(self): from sapcli.auth import get_password, set_password with patch("sapcli.password._KEYRING_AVAILABLE", False): self.assertIsNone(get_password("h", "c", "u")) self.assertFalse(set_password("h", "c", "u", "p")) # ═══════════════════════════════════════════ # TestXmlInjection # ═══════════════════════════════════════════ class TestXmlInjection(unittest.TestCase): def test_xml_injection_in_ddic(self): from sapcli.ddic import DomainDefinition evil = 'ZTEST<>&"' d = DomainDefinition(datatype="CHAR", length=10) xml = d.to_xml(evil, evil) self.assertNotIn('name="ZTEST<', xml) self.assertIn("<", xml) self.assertIn(">", xml) def test_xml_injection_in_ddl(self): from sapcli.ddic import TableDefinition, TableField evil_desc = "O'Brien's \"test\"" t = TableDefinition(fields=[TableField(name="ID", type_name="Z_ID")]) ddl = t.to_ddl("ZMY_TABLE", evil_desc) self.assertIn("O''Brien", ddl) def test_empty_name_handling(self): from sapcli.ddic import DomainDefinition d = DomainDefinition(datatype="CHAR", length=10) xml = d.to_xml("", "") self.assertIsInstance(xml, str) # ═══════════════════════════════════════════ # TestClientMethods # ═══════════════════════════════════════════ class TestClientMethods(unittest.TestCase): def _make_client(self): from sapcli.client import ADTClient client = ADTClient.__new__(ADTClient) client.host = "http://test:8000" client.csrf_token = "test-token" client.sap_client = "100" client.session = MagicMock() client._stateful = False return client def test_list_objects(self): client = self._make_client() mock_resp = MagicMock() mock_resp.status_code = 200 mock_resp.content = ( b'' b'' b'' b'' b'' b'' ) client.session.get.return_value = mock_resp result = client.list_objects(obj_type="report") self.assertIsInstance(result, list) self.assertEqual(len(result), 1) self.assertEqual(result[0]["name"], "Z_TEST") def test_transport_info(self): client = self._make_client() mock_resp = MagicMock() mock_resp.status_code = 200 mock_resp.content = ( b'' b'' ) client.session.get.return_value = mock_resp result = client.transport_info("DEVK901362") self.assertIsInstance(result, dict) def test_pretty_print(self): client = self._make_client() mock_resp = MagicMock() mock_resp.status_code = 200 mock_resp.text = 'FORMATTED CODE' client.session.post.return_value = mock_resp result = client.pretty_print("raw code") self.assertIsInstance(result, str) # ═══════════════════════════════════════════ # TestAnalyze # ═══════════════════════════════════════════ class TestAnalyze(unittest.TestCase): def test_analyze_type_ref(self): from sapcli.commands.analyze import _analyze_dependencies code = 'DATA: lo_obj TYPE REF TO zcl_my_class.\nDATA: lo_if TYPE REF TO zif_my_interface.\n' deps = _analyze_dependencies(code) self.assertIn("ZCL_MY_CLASS", deps) def test_analyze_call_function(self): from sapcli.commands.analyze import _analyze_dependencies code = "CALL FUNCTION 'Z_MY_FUNC'\n EXPORTING iv_param = lv_val.\n" deps = _analyze_dependencies(code) self.assertIn("Z_MY_FUNC", deps) def test_analyze_select_table(self): from sapcli.commands.analyze import _analyze_dependencies code = "SELECT * FROM zmy_table INTO TABLE @lt_data.\n" deps = _analyze_dependencies(code) self.assertIn("ZMY_TABLE", deps) def test_analyze_builtin_filtered(self): from sapcli.commands.analyze import _analyze_dependencies code = "DATA: lv_str TYPE string.\nDATA: lv_int TYPE i.\n" deps = _analyze_dependencies(code) self.assertNotIn("STRING", deps) self.assertNotIn("I", deps) # ═══════════════════════════════════════════ # TestScaffold # ═══════════════════════════════════════════ class TestScaffold(unittest.TestCase): def test_scaffold_alv_report(self): from sapcli.commands.scaffold import _alv_report_template result = _alv_report_template("ZMY_REPORT", "$TMP") self.assertIn("ZMY_REPORT", result) self.assertIn("cl_salv_table", result) def test_scaffold_interface_class(self): from sapcli.commands.scaffold import _interface_class_template result = _interface_class_template("ZCL_MY_CLASS", "$TMP") self.assertIn("ZCL_MY_CLASS", result) self.assertIn("INTERFACE", result) def test_scaffold_data_model(self): from sapcli.commands.scaffold import _data_model_template result = _data_model_template("ZMY_MODEL", "$TMP") self.assertIn("ZMY_MODEL", result) self.assertIn("define table", result) def test_scaffold_bapi_wrapper(self): from sapcli.commands.scaffold import _bapi_wrapper_template result = _bapi_wrapper_template("Z_MY_BAPI", "$TMP") self.assertIn("Z_MY_BAPI", result) self.assertIn("CLASS", result) # ═══════════════════════════════════════════ # TestOutput # ═══════════════════════════════════════════ class TestOutput(unittest.TestCase): def test_print_header(self): from sapcli.cli.output import print_header captured = io.StringIO() with patch("sys.stdout", captured): print_header("测试标题") self.assertIn("测试标题", captured.getvalue()) def test_print_success(self): from sapcli.cli.output import print_success captured = io.StringIO() with patch("sys.stdout", captured): print_success("操作成功") self.assertIn("✓", captured.getvalue()) def test_print_error(self): from sapcli.cli.output import print_error captured = io.StringIO() with patch("sys.stdout", captured): print_error("操作失败") self.assertIn("✗", captured.getvalue()) # ═══════════════════════════════════════════ # TestDiff # ═══════════════════════════════════════════ class TestDiff(unittest.TestCase): def test_unified_diff_logic(self): """验证 difflib.unified_diff 逻辑(与 diff_cmd.py 一致)。""" import difflib old = "LINE 1\nLINE 2\nLINE 3\n".splitlines(keepends=True) new = "LINE 1\nLINE 2 MODIFIED\nLINE 3\n".splitlines(keepends=True) diff = list(difflib.unified_diff(old, new, fromfile="SAP", tofile="LOCAL", lineterm="")) self.assertTrue(any("LINE 2" in l for l in diff)) self.assertTrue(any(l.startswith("-") and "LINE 2" in l and not l.startswith("---") for l in diff)) self.assertTrue(any(l.startswith("+") and "MODIFIED" in l and not l.startswith("+++") for l in diff)) # ═══════════════════════════════════════════ # TestOpenSpec # ═══════════════════════════════════════════ class TestOpenSpec(unittest.TestCase): def test_openspec_structure_exists(self): base = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.dirname(base) self.assertTrue(os.path.isdir(os.path.join(project_root, "openspec"))) def test_openspec_config_exists(self): base = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.dirname(base) self.assertTrue(os.path.isfile(os.path.join(project_root, "openspec", "config.yaml"))) def test_openspec_specs_exist(self): base = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.dirname(base) specs_dir = os.path.join(project_root, "openspec", "specs") self.assertTrue(os.path.isdir(specs_dir)) # Source of Truth — 7 domains describing current system behavior for domain in ["connection", "object-lifecycle", "batch-ops", "search-browse", "transport", "quality", "config"]: self.assertTrue(os.path.isfile(os.path.join(specs_dir, domain, "spec.md"))) # ═══════════════════════════════════════════ # TestCLAUDEmd # ═══════════════════════════════════════════ class TestCLAUDEmd(unittest.TestCase): def test_claude_md_exists(self): base = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.dirname(base) self.assertTrue(os.path.isfile(os.path.join(project_root, "docs", "dev", "CLAUDE.md"))) def test_agents_md_exists(self): base = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.dirname(base) self.assertTrue(os.path.isfile(os.path.join(project_root, "docs", "dev", "AGENTS.md"))) # ═══════════════════════════════════════════ # TestModuleStructure # ═══════════════════════════════════════════ class TestModuleStructure(unittest.TestCase): def test_cli_package(self): import sapcli.cli import sapcli.cli.parser import sapcli.cli.output def test_commands_package(self): import sapcli.commands import sapcli.commands.crud import sapcli.commands.batch import sapcli.commands.config_cmd import sapcli.commands.search import sapcli.commands.transport import sapcli.commands.quality import sapcli.commands.diff_cmd import sapcli.commands.package_cmd import sapcli.commands.cds import sapcli.commands.analyze import sapcli.commands.scaffold def test_utils_package(self): import sapcli.utils import sapcli.utils.xml_utils def test_auth_module(self): import sapcli.auth def test_all_commands_importable(self): from sapcli.commands import ( cmd_create, cmd_delete, cmd_download, cmd_info, cmd_sync, cmd_init, cmd_refresh, cmd_sync_all, cmd_config, cmd_list, cmd_whereused, cmd_search, cmd_transport, cmd_check, cmd_format, cmd_diff, cmd_package, cmd_cds, cmd_analyze, cmd_scaffold, ) # ═══════════════════════════════════════════ if __name__ == "__main__": sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace") unittest.main(verbosity=2)