sap-cli-skill/tests/test_e2e.md

""" 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)