""" sap-cli 端到端测试 — 使用真实项目目录 + mock SAP 验证完整批量同步流程: init → dry-run → sync --all → refresh 运行方式: python test/script/test_e2e.py """ from __future__ import annotations import io import json import os import shutil import sys import tempfile import unittest from unittest.mock import MagicMock # Windows UTF-8 if sys.platform == "win32": sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace") PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) E2E_PROJECT = os.path.join(PROJECT_ROOT, "test", "e2e_project") class TestE2EInit(unittest.TestCase): """端到端: init 命令。""" def setUp(self): # 复制测试项目到临时目录，避免污染原始文件 self.tmpdir = tempfile.mkdtemp(prefix="sapcli_e2e_") shutil.copytree(E2E_PROJECT, self.tmpdir, dirs_exist_ok=True) self.client = self._mock_client() def tearDown(self): shutil.rmtree(self.tmpdir, ignore_errors=True) def _mock_client(self): client = MagicMock() client.get_object_status.return_value = { "exists": True, "status": "active", "corr_nr": "DEVK901362" } return client def test_init_generates_manifest(self): """init 正确生成清单，包含所有扫描到的对象。""" import argparse from sapcli.commands import cmd_init from sapcli.manifest import Manifest args = argparse.Namespace(path=self.tmpdir) cmd_init(args, self.client) manifest = Manifest.load(self.tmpdir) # 验证扫描到的对象 self.assertEqual(len(manifest.objects), 6) self.assertIn("ZHELLO", manifest.objects) self.assertIn("ZSAPILOT_OBJ", manifest.objects) self.assertIn("ZCL_SAPILOT_CONSTANTS", manifest.objects) self.assertIn("ZCL_SAPILOT_HTTP_HANDLER", manifest.objects) self.assertIn("ZCL_SAPILOT_OP_ADMIN", manifest.objects) self.assertIn("ZGROUP/Z_MY_FUNC", manifest.objects) # 验证类型映射 self.assertEqual(manifest.get("ZHELLO").type, "report") self.assertEqual(manifest.get("ZCL_SAPILOT_CONSTANTS").type, "class") self.assertEqual(manifest.get("ZGROUP/Z_MY_FUNC").type, "function") # 验证相对路径 self.assertEqual(manifest.get("ZHELLO").file, "reports/zhello.abap") self.assertEqual(manifest.get("ZGROUP/Z_MY_FUNC").file, "functions/zgroup/z_my_func.abap") def test_init_manifest_json_structure(self): """init 生成的 manifest.json 结构正确。""" import argparse from sapcli.commands import cmd_init args = argparse.Namespace(path=self.tmpdir) cmd_init(args, self.client) with open(os.path.join(self.tmpdir, "manifest.json"), "r", encoding="utf-8") as f: data = json.load(f) self.assertEqual(data["version"], 1) self.assertIsNotNone(data["last_init"]) self.assertIsNone(data["last_refresh"]) self.assertEqual(len(data["objects"]), 6) # 验证单个对象结构 zhello = data["objects"]["ZHELLO"] self.assertEqual(zhello["type"], "report") self.assertEqual(zhello["file"], "reports/zhello.abap") self.assertEqual(zhello["system_status"], "active") self.assertEqual(zhello["corr_nr"], "DEVK901362") self.assertEqual(zhello["depends_on"], []) self.assertIsNone(zhello["last_sync"]) self.assertEqual(zhello["last_sync_result"], "pending") def test_init_mixed_statuses(self): """init 正确处理混合 SAP 状态（存在/不存在/未激活）。""" import argparse from sapcli.commands import cmd_init from sapcli.manifest import Manifest def mock_status(uri): if "zhello" in uri: return {"exists": True, "status": "active", "corr_nr": "DEVK901362"} elif "zsapilot_obj" in uri: return {"exists": True, "status": "inactive", "corr_nr": None} else: return {"exists": False, "status": "not_exists", "corr_nr": None} self.client.get_object_status.side_effect = mock_status args = argparse.Namespace(path=self.tmpdir) cmd_init(args, self.client) manifest = Manifest.load(self.tmpdir) self.assertEqual(manifest.get("ZHELLO").system_status, "active") self.assertEqual(manifest.get("ZHELLO").corr_nr, "DEVK901362") self.assertEqual(manifest.get("ZSAPILOT_OBJ").system_status, "inactive") self.assertIsNone(manifest.get("ZSAPILOT_OBJ").corr_nr) self.assertEqual(manifest.get("ZCL_SAPILOT_CONSTANTS").system_status, "not_exists") class TestE2ESyncAll(unittest.TestCase): """端到端: sync --all 命令。""" def setUp(self): self.tmpdir = tempfile.mkdtemp(prefix="sapcli_e2e_") shutil.copytree(E2E_PROJECT, self.tmpdir, dirs_exist_ok=True) self.client = self._mock_client() def tearDown(self): shutil.rmtree(self.tmpdir, ignore_errors=True) def _mock_client(self): client = MagicMock() client.object_exists.return_value = True client.lock.return_value = ("handle", "DEVK901362") client.unlock.return_value = True client.set_source.return_value = True client.syntax_check.return_value = (True, []) client.activate.return_value = (True, []) client.get_object_status.return_value = { "exists": True, "status": "active", "corr_nr": "DEVK901362" } return client def _create_manifest(self): """先执行 init 创建清单。""" import argparse from sapcli.commands import cmd_init args = argparse.Namespace(path=self.tmpdir) cmd_init(args, self.client) def test_dry_run_no_side_effects(self): """dry-run 不修改清单状态。""" import argparse from sapcli.commands import cmd_sync_all from sapcli.manifest import Manifest self._create_manifest() args = argparse.Namespace(path=self.tmpdir, dry_run=True, fail_fast=False) cmd_sync_all(args, self.client) manifest = Manifest.load(self.tmpdir) # 所有对象应保持 pending 状态 for entry in manifest.objects.values(): self.assertEqual(entry.last_sync_result, "pending") # activate 不应被调用 self.client.activate.assert_not_called() def test_sync_all_updates_manifest(self): """sync --all 成功后所有对象标记为 success。""" import argparse from sapcli.commands import cmd_sync_all from sapcli.manifest import Manifest self._create_manifest() args = argparse.Namespace(path=self.tmpdir, dry_run=False, fail_fast=False) cmd_sync_all(args, self.client) manifest = Manifest.load(self.tmpdir) for name, entry in manifest.objects.items(): self.assertEqual(entry.system_status, "active", f"{name} 应为 active") self.assertEqual(entry.last_sync_result, "success", f"{name} 应为 success") self.assertIsNotNone(entry.last_sync, f"{name} 应有 last_sync") def test_sync_all_respects_execution_order(self): """sync --all 按类型优先级执行。""" import argparse from sapcli.commands import cmd_sync_all self._create_manifest() args = argparse.Namespace(path=self.tmpdir, dry_run=False, fail_fast=False) cmd_sync_all(args, self.client) # 验证 activate 的调用顺序 calls = self.client.activate.call_args_list names = [call[0][0] for call in calls] # 第一个位置参数是 name # interface 应在 class 前，class 应在 report 前 class_idx = next((i for i, n in enumerate(names) if "ZCL_SAPILOT" in n), None) report_idx = next((i for i, n in enumerate(names) if "ZHELLO" in n or "ZSAPILOT" in n), None) if class_idx is not None and report_idx is not None: self.assertLess(class_idx, report_idx, "class 应在 report 之前执行") def test_sync_all_with_dependency_failure(self): """sync --all 某对象失败时依赖它的对象被跳过。""" import argparse from sapcli.commands import cmd_sync_all from sapcli.manifest import Manifest, ManifestEntry, init_manifest # 手动创建带依赖的清单 m = init_manifest(self.tmpdir) m.upsert(ManifestEntry( name="ZCL_SAPILOT_CONSTANTS", type="class", file="classes/zcl_sapilot_constants.abap", system_status="inactive", last_sync_result="pending", )) m.upsert(ManifestEntry( name="ZSAPILOT_OBJ", type="report", file="reports/zsapilot_obj.abap", system_status="inactive", last_sync_result="pending", depends_on=["ZCL_SAPILOT_CONSTANTS"], )) m.save() # 让 class 的 activate 失败 def mock_activate(name, uri, corr_nr=None): if "zcl_sapilot_constants" in name.lower(): return (False, [{"type": "E", "line": "1", "text": "test error", "href": ""}]) return (True, []) self.client.activate.side_effect = mock_activate args = argparse.Namespace(path=self.tmpdir, dry_run=False, fail_fast=False) cmd_sync_all(args, self.client) manifest = Manifest.load(self.tmpdir) self.assertEqual(manifest.get("ZCL_SAPILOT_CONSTANTS").last_sync_result, "failed") self.assertEqual(manifest.get("ZSAPILOT_OBJ").last_sync_result, "skipped") class TestE2ERefresh(unittest.TestCase): """端到端: refresh 命令。""" def setUp(self): self.tmpdir = tempfile.mkdtemp(prefix="sapcli_e2e_") shutil.copytree(E2E_PROJECT, self.tmpdir, dirs_exist_ok=True) self.client = MagicMock() def tearDown(self): shutil.rmtree(self.tmpdir, ignore_errors=True) def test_refresh_preserves_sync_history(self): """refresh 只更新 system_status 和 corr_nr，不动 last_sync。""" import argparse from sapcli.commands import cmd_refresh from sapcli.manifest import Manifest, ManifestEntry, init_manifest # 创建带同步历史的清单 m = init_manifest(self.tmpdir) m.upsert(ManifestEntry( name="ZCL_SAPILOT_CONSTANTS", type="class", file="classes/zcl_sapilot_constants.abap", system_status="active", corr_nr="DEVK901362", last_sync="2026-01-01T00:00:00", last_sync_result="success", )) m.save() # SAP 端状态变更 self.client.get_object_status.return_value = { "exists": True, "status": "inactive", "corr_nr": "DEVK901363" } args = argparse.Namespace(path=self.tmpdir) cmd_refresh(args, self.client) manifest = Manifest.load(self.tmpdir) entry = manifest.get("ZCL_SAPILOT_CONSTANTS") # 状态已更新 self.assertEqual(entry.system_status, "inactive") self.assertEqual(entry.corr_nr, "DEVK901363") # 同步历史不变 self.assertEqual(entry.last_sync, "2026-01-01T00:00:00") self.assertEqual(entry.last_sync_result, "success") # ════════════════════════════════════════════════════════════════ # 运行 # ════════════════════════════════════════════════════════════════ if __name__ == "__main__": print("=" * 60) print(" sap-cli 端到端测试 — 真实项目目录 + mock SAP") print("=" * 60) print(f" 测试项目: {E2E_PROJECT}") loader = unittest.TestLoader() suite = unittest.TestSuite() for cls in [TestE2EInit, TestE2ESyncAll, TestE2ERefresh]: suite.addTests(loader.loadTestsFromTestCase(cls)) runner = unittest.TextTestRunner(verbosity=2) result = runner.run(suite) print("\n" + "=" * 60) if result.wasSuccessful(): print(f" ✓ 全部通过! ({result.testsRun} 个测试)") else: print(f" ✗ 失败: {len(result.failures)} 错误: {len(result.errors)}") print("=" * 60) sys.exit(0 if result.wasSuccessful() else 1)