吴让宇
c5905a5b1e
方向反转：此前 SKILL.md 是「模板渲染产物」、sap-cli 是源；现 sap-cli 归档， sap-cli-skill 承接开发与分发，SKILL.md 回归手工维护的正本。 迁移（来自 sap-cli，共 104 文件）： - tests/ 692 例测试（15 个文件的内联 sys.path 改指 assets/） - openspec/ SDD 规格与归档变更（42 文件） - docs/ 开发文档与 ADT 原理（含 dev/CLAUDE.md、AGENTS.md） - .claude/ rules 副本 + settings.json（供 Claude Code） - .github/ .hermes/ .pre-commit-config.yaml .editorconfig CLAUDE.md - scripts/ 保持仅 setup.py（pack_skill.py 已随旧仓归档，不迁） 修复（迁移暴露的真实缺陷）： - assets/pyproject.toml 的 build-backend 写作 `setuptools.backends._legacy:_Backend`， 该模块在 setuptools 中不存在 → `pip install -e` 从来装不上。改为 build_meta。 实测：临时 venv 安装成功，sap-cli --help 正常列出 31 个命令 - pyproject readme 指向不存在的 assets/README.md（editable 安装会失败）→ 改内联文本 - pyproject urls 改指 sap-cli-skill 机制调整： - .github/workflows/ci.yml 适配 assets/ 布局；顶部注明该工作流仅 GitHub 执行， 本仓在 Gitee 不会自动跑 - pre-commit 增本地测试门禁（Gitee 上真正生效的那道） - .gitignore 合并旧仓完整规则（保留 log/ 下 md 知识库入库，只忽略运行日志） - 大文件上限 100KB→1MB（架构图 512KB） 守卫测试 tests/unit/test_repo_guards.py（10 → 18 例）： - SKILL.md 须记录 parser 全部 CLI 命令 / 铁律 1-5 须为真实小节标题 / 示例不得违反铁律 5 - references/ 规则齐备；.claude/rules 与 references 必须一致（实测抓到一次真实漂移） - VERSION == sapcli.__version__ == README 版本 - 仓内不得再出现 pack_skill.py / skill-src（防废弃流程回潮） 698 tests OK；editable 安装与 CLI 入口经临时 venv 实测通过。 docs/RELEASING.md 重写为单源开发流程。
333 lines
12 KiB
Markdown
333 lines
12 KiB
Markdown
"""
|
|
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)
|