吴让宇
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 重写为单源开发流程。
734 lines
30 KiB
Python
734 lines
30 KiB
Python
"""batch.py / analyze.py 深入单元测试 — 失败/边界场景。
|
||
|
||
运行: python tests/unit/test_batch_analyze.py
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from unittest.mock import MagicMock, patch, call
|
||
|
||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "assets"))
|
||
|
||
from sapcli.manifest import Manifest, ManifestEntry
|
||
|
||
|
||
# ─── Helpers ───
|
||
|
||
def _make_client():
|
||
"""创建完整 mock 的 ADTClient。"""
|
||
client = MagicMock()
|
||
client.host = "https://sap.example.com"
|
||
client.sap_client = "100"
|
||
client.csrf_token = "test-csrf-token"
|
||
client.user = "TESTUSER"
|
||
client.session = MagicMock()
|
||
return client
|
||
|
||
|
||
def _args(**kwargs):
|
||
defaults = {"type": "report", "name": "ZTEST", "path": ".", "config": None}
|
||
defaults.update(kwargs)
|
||
return argparse.Namespace(**defaults)
|
||
|
||
|
||
def _write_manifest(td, entries, version=1):
|
||
"""在 td 下写 manifest.json。entries: list[ManifestEntry]。"""
|
||
data = {
|
||
"version": version,
|
||
"objects": {e.name: e.to_dict() for e in entries},
|
||
}
|
||
path = os.path.join(td, "manifest.json")
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f)
|
||
return path
|
||
|
||
|
||
def _make_entry(name="ZTEST", type_="report", file_="reports/ztest.abap",
|
||
status="not_exists", corr_nr=None, depends_on=None,
|
||
last_sync=None, last_sync_result="pending"):
|
||
return ManifestEntry(
|
||
name=name,
|
||
type=type_,
|
||
file=file_,
|
||
system_status=status,
|
||
corr_nr=corr_nr,
|
||
depends_on=depends_on or [],
|
||
last_sync=last_sync,
|
||
last_sync_result=last_sync_result,
|
||
)
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# batch.py — cmd_init
|
||
# ═══════════════════════════════════════════
|
||
|
||
class TestCmdInitEmptyDir(unittest.TestCase):
|
||
"""cmd_init 无 .abap 文件的空目录。"""
|
||
|
||
def test_no_abap_files_returns_early(self):
|
||
from sapcli.commands.batch import cmd_init
|
||
client = _make_client()
|
||
with tempfile.TemporaryDirectory() as td:
|
||
# 空目录，没有任何子目录或 .abap 文件
|
||
cmd_init(_args(path=td), client)
|
||
# 不应创建 manifest.json
|
||
self.assertFalse(os.path.isfile(os.path.join(td, "manifest.json")))
|
||
# client 不应被调用
|
||
client.get_object_status.assert_not_called()
|
||
|
||
def test_no_abap_files_with_subdirs(self):
|
||
"""有标准子目录但无 .abap 文件。"""
|
||
from sapcli.commands.batch import cmd_init
|
||
client = _make_client()
|
||
with tempfile.TemporaryDirectory() as td:
|
||
for d in ("reports", "classes", "interfaces"):
|
||
os.makedirs(os.path.join(td, d))
|
||
cmd_init(_args(path=td), client)
|
||
self.assertFalse(os.path.isfile(os.path.join(td, "manifest.json")))
|
||
client.get_object_status.assert_not_called()
|
||
|
||
|
||
class TestCmdInitDirNotExists(unittest.TestCase):
|
||
"""cmd_init 目录不存在。"""
|
||
|
||
def test_nonexistent_dir_raises_config_error(self):
|
||
from sapcli.commands.batch import cmd_init
|
||
from sapcli.exceptions import ConfigError
|
||
client = _make_client()
|
||
with self.assertRaises(ConfigError) as ctx:
|
||
cmd_init(_args(path="/nonexistent/path/xyz"), client)
|
||
self.assertIn("不存在", str(ctx.exception))
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# batch.py — cmd_sync_all
|
||
# ═══════════════════════════════════════════
|
||
|
||
class TestCmdSyncAllNoManifest(unittest.TestCase):
|
||
"""cmd_sync_all 无清单文件报错。"""
|
||
|
||
def test_missing_manifest_raises_file_not_found(self):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
client = _make_client()
|
||
with tempfile.TemporaryDirectory() as td:
|
||
# 不创建 manifest.json
|
||
with self.assertRaises(FileNotFoundError):
|
||
cmd_sync_all(_args(path=td), client)
|
||
|
||
|
||
class TestCmdSyncAllDryRun(unittest.TestCase):
|
||
"""cmd_sync_all dry-run 模式：只打印计划，不实际同步。"""
|
||
|
||
def test_dry_run_does_not_sync(self):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
client = _make_client()
|
||
with tempfile.TemporaryDirectory() as td:
|
||
# 创建源文件
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
src_file = os.path.join(report_dir, "ztest.abap")
|
||
with open(src_file, "w", encoding="utf-8") as f:
|
||
f.write("REPORT ztest.")
|
||
|
||
entry = _make_entry(status="inactive", last_sync_result="failed")
|
||
_write_manifest(td, [entry])
|
||
|
||
with patch("builtins.print") as mock_print:
|
||
cmd_sync_all(_args(path=td, dry_run=True), client)
|
||
|
||
# _sync_single 不应被调用
|
||
# 检查打印了 dry-run 标题
|
||
printed = "\n".join(str(c) for c in mock_print.call_args_list)
|
||
self.assertIn("dry-run", printed)
|
||
|
||
def test_dry_run_shows_pending_objects(self):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
client = _make_client()
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
with open(os.path.join(report_dir, "ztest.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT ztest.")
|
||
with open(os.path.join(report_dir, "zother.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT zother.")
|
||
|
||
e1 = _make_entry(name="ZTEST", file_="reports/ztest.abap", status="not_exists")
|
||
e2 = _make_entry(name="ZOTHER", file_="reports/zother.abap", status="inactive")
|
||
_write_manifest(td, [e1, e2])
|
||
|
||
with patch("builtins.print") as mock_print:
|
||
cmd_sync_all(_args(path=td, dry_run=True), client)
|
||
|
||
printed = "\n".join(str(c) for c in mock_print.call_args_list)
|
||
self.assertIn("ZTEST", printed)
|
||
self.assertIn("ZOTHER", printed)
|
||
|
||
|
||
class TestCmdSyncAllFailFast(unittest.TestCase):
|
||
"""cmd_sync_all 有对象同步失败（fail-fast）。"""
|
||
|
||
@patch("sapcli.commands.batch._sync_single")
|
||
def test_fail_fast_stops_on_first_error(self, mock_sync):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
mock_sync.return_value = (False, "同步失败: 语法错误", None)
|
||
client = _make_client()
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
|
||
e1 = _make_entry(name="ZFIRST", file_="reports/zfirst.abap", status="inactive")
|
||
e2 = _make_entry(name="ZSECOND", file_="reports/zsecond.abap", status="inactive")
|
||
_write_manifest(td, [e1, e2])
|
||
|
||
for fname in ("zfirst.abap", "zsecond.abap"):
|
||
with open(os.path.join(report_dir, fname), "w", encoding="utf-8") as f:
|
||
f.write("REPORT ztest.")
|
||
|
||
with patch("builtins.print"):
|
||
cmd_sync_all(_args(path=td, fail_fast=True), client)
|
||
|
||
# fail_fast → 第一个失败后 break，_sync_single 最多调用 1 次
|
||
self.assertLessEqual(mock_sync.call_count, 1)
|
||
|
||
@patch("sapcli.commands.batch._sync_single")
|
||
def test_no_fail_fast_continues_on_error(self, mock_sync):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
# 第一次失败，第二次成功
|
||
mock_sync.side_effect = [
|
||
(False, "同步失败", None),
|
||
(True, None, "DEVK001"),
|
||
]
|
||
client = _make_client()
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
|
||
e1 = _make_entry(name="ZFIRST", file_="reports/zfirst.abap", status="inactive")
|
||
e2 = _make_entry(name="ZSECOND", file_="reports/zsecond.abap", status="inactive")
|
||
_write_manifest(td, [e1, e2])
|
||
|
||
for fname in ("zfirst.abap", "zsecond.abap"):
|
||
with open(os.path.join(report_dir, fname), "w", encoding="utf-8") as f:
|
||
f.write("REPORT ztest.")
|
||
|
||
with patch("builtins.print"):
|
||
cmd_sync_all(_args(path=td, fail_fast=False), client)
|
||
|
||
# 没有 fail_fast → 继续同步第二个
|
||
self.assertEqual(mock_sync.call_count, 2)
|
||
|
||
@patch("sapcli.commands.batch._sync_single")
|
||
def test_cascading_skip_on_dep_failure(self, mock_sync):
|
||
"""依赖失败时，后续依赖它的对象被级联跳过。"""
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
mock_sync.return_value = (False, "同步失败", None)
|
||
client = _make_client()
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
|
||
e1 = _make_entry(name="ZBASE", file_="reports/zbase.abap", status="inactive")
|
||
e2 = _make_entry(name="ZCHILD", file_="reports/zchild.abap", status="inactive",
|
||
depends_on=["ZBASE"])
|
||
_write_manifest(td, [e1, e2])
|
||
|
||
for fname in ("zbase.abap", "zchild.abap"):
|
||
with open(os.path.join(report_dir, fname), "w", encoding="utf-8") as f:
|
||
f.write("REPORT ztest.")
|
||
|
||
with patch("builtins.print") as mock_print:
|
||
cmd_sync_all(_args(path=td, fail_fast=False), client)
|
||
|
||
printed = "\n".join(str(c) for c in mock_print.call_args_list)
|
||
# ZCHILD 应被跳过（依赖失败）
|
||
self.assertIn("ZCHILD", printed)
|
||
self.assertIn("跳过", printed)
|
||
|
||
|
||
class TestCmdSyncAllMissingFile(unittest.TestCase):
|
||
"""cmd_sync_all 清单中对象对应的本地文件缺失。"""
|
||
|
||
@patch("sapcli.commands.batch._sync_single")
|
||
def test_missing_file_is_skipped(self, mock_sync):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
mock_sync.return_value = (True, None, "DEVK001")
|
||
client = _make_client()
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
# 不创建源文件 → 清单中 file 指向不存在的文件
|
||
e1 = _make_entry(name="ZMISSING", file_="reports/zmissing.abap", status="inactive")
|
||
_write_manifest(td, [e1])
|
||
|
||
with patch("builtins.print"):
|
||
cmd_sync_all(_args(path=td), client)
|
||
|
||
# _sync_single 不应被调用（文件缺失，对象被跳过）
|
||
mock_sync.assert_not_called()
|
||
|
||
|
||
class TestCmdSyncAllAlreadySynced(unittest.TestCase):
|
||
"""cmd_sync_all 跳过已同步且激活的对象。"""
|
||
|
||
@patch("sapcli.commands.batch._sync_single")
|
||
def test_active_success_is_skipped(self, mock_sync):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
client = _make_client()
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
with open(os.path.join(report_dir, "zactive.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT zactive.")
|
||
|
||
e1 = _make_entry(
|
||
name="ZACTIVE", file_="reports/zactive.abap",
|
||
status="active", last_sync_result="success",
|
||
last_sync="2026-01-01T00:00:00+00:00",
|
||
)
|
||
_write_manifest(td, [e1])
|
||
|
||
with patch("builtins.print"):
|
||
cmd_sync_all(_args(path=td), client)
|
||
|
||
mock_sync.assert_not_called()
|
||
|
||
|
||
class TestCmdSyncAllCyclicDependency(unittest.TestCase):
|
||
"""cmd_sync_all 拓扑排序检测到循环依赖时抛异常。"""
|
||
|
||
def test_cyclic_raises(self):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
from sapcli.exceptions import CyclicDependencyError
|
||
client = _make_client()
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
for fname in ("za.abap", "zb.abap"):
|
||
with open(os.path.join(report_dir, fname), "w", encoding="utf-8") as f:
|
||
f.write("REPORT ztest.")
|
||
|
||
e1 = _make_entry(name="ZA", file_="reports/za.abap", status="inactive",
|
||
depends_on=["ZB"])
|
||
e2 = _make_entry(name="ZB", file_="reports/zb.abap", status="inactive",
|
||
depends_on=["ZA"])
|
||
_write_manifest(td, [e1, e2])
|
||
|
||
with self.assertRaises(CyclicDependencyError):
|
||
cmd_sync_all(_args(path=td), client)
|
||
|
||
|
||
class TestCmdSyncAllNotExistsAutoCreate(unittest.TestCase):
|
||
"""cmd_sync_all 对象不存在时自动创建，创建失败则 fail-fast 或 continue。"""
|
||
|
||
def test_auto_create_failure_fail_fast(self):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
client = _make_client()
|
||
client.create_object.side_effect = Exception("SAP 连接超时")
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
with open(os.path.join(report_dir, "znew.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT znew.")
|
||
|
||
e1 = _make_entry(name="ZNEW", file_="reports/znew.abap", status="not_exists")
|
||
_write_manifest(td, [e1])
|
||
|
||
with patch("builtins.print"):
|
||
cmd_sync_all(_args(path=td, fail_fast=True), client)
|
||
|
||
client.create_object.assert_called_once()
|
||
|
||
def test_auto_create_failure_continue(self):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
client = _make_client()
|
||
client.create_object.side_effect = Exception("SAP 连接超时")
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
with open(os.path.join(report_dir, "znew.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT znew.")
|
||
|
||
e1 = _make_entry(name="ZNEW", file_="reports/znew.abap", status="not_exists")
|
||
_write_manifest(td, [e1])
|
||
|
||
with patch("builtins.print"):
|
||
# 不应抛出异常，只是标记为 failed
|
||
cmd_sync_all(_args(path=td, fail_fast=False), client)
|
||
|
||
# 验证 manifest 被更新了（即使失败也保存）
|
||
self.assertTrue(os.path.isfile(os.path.join(td, "manifest.json")))
|
||
|
||
|
||
class TestCmdSyncAllFunctionAutoGroup(unittest.TestCase):
|
||
"""cmd_sync_all function 类型自动创建函数组。"""
|
||
|
||
def test_auto_creates_function_group(self):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
client = _make_client()
|
||
client.function_group_exists.return_value = False
|
||
client.create_function_group.return_value = True
|
||
# create_object for the function module itself → succeed
|
||
client.create_object.return_value = ("/uri/zgrp/zfunc", "/uri/zgrp/zfunc/source/main")
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
func_dir = os.path.join(td, "functions", "zgrp")
|
||
os.makedirs(func_dir)
|
||
with open(os.path.join(func_dir, "zfunc.abap"), "w", encoding="utf-8") as f:
|
||
f.write("FUNCTION zfunc.")
|
||
|
||
e1 = _make_entry(
|
||
name="ZGRP/ZFUNC", type_="function",
|
||
file_="functions/zgrp/zfunc.abap", status="not_exists",
|
||
)
|
||
_write_manifest(td, [e1])
|
||
|
||
with patch("sapcli.commands.batch._sync_single") as mock_sync, \
|
||
patch("builtins.print"):
|
||
# _sync_single succeeds
|
||
mock_sync.return_value = (True, None, "DEVK001")
|
||
cmd_sync_all(_args(path=td), client)
|
||
|
||
client.create_function_group.assert_called_once_with("ZGRP")
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# batch.py — cmd_init SAP 查询失败
|
||
# ═══════════════════════════════════════════
|
||
|
||
class TestCmdInitSAPQueryFailure(unittest.TestCase):
|
||
"""cmd_init 查询 SAP 状态失败时使用 fallback 状态。"""
|
||
|
||
def test_query_failure_defaults_to_not_exists(self):
|
||
from sapcli.commands.batch import cmd_init
|
||
client = _make_client()
|
||
client.get_object_status.side_effect = Exception("网络超时")
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
with open(os.path.join(report_dir, "ztest.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT ztest.")
|
||
|
||
with patch("builtins.print"):
|
||
cmd_init(_args(path=td), client)
|
||
|
||
# 应仍然创建 manifest.json
|
||
manifest_path = os.path.join(td, "manifest.json")
|
||
self.assertTrue(os.path.isfile(manifest_path))
|
||
|
||
with open(manifest_path, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
# 失败时 fallback 到 not_exists
|
||
self.assertEqual(data["objects"]["ZTEST"]["system_status"], "not_exists")
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# batch.py — cmd_refresh 查询失败
|
||
# ═══════════════════════════════════════════
|
||
|
||
class TestCmdRefreshQueryFailure(unittest.TestCase):
|
||
"""cmd_refresh 单个对象查询失败时跳过。"""
|
||
|
||
def test_query_failure_skips_entry(self):
|
||
from sapcli.commands.batch import cmd_refresh
|
||
client = _make_client()
|
||
client.get_object_status.side_effect = Exception("SAP 不可用")
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
entry = _make_entry(status="active")
|
||
_write_manifest(td, [entry])
|
||
|
||
with patch("builtins.print"):
|
||
cmd_refresh(_args(path=td), client)
|
||
|
||
# 不抛异常，只跳过失败的对象
|
||
client.get_object_status.assert_called()
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# analyze.py — cmd_analyze
|
||
# ═══════════════════════════════════════════
|
||
|
||
class TestCmdAnalyzeDirNotExists(unittest.TestCase):
|
||
"""cmd_analyze 目录不存在。"""
|
||
|
||
def test_nonexistent_dir_prints_error(self):
|
||
from sapcli.commands.analyze import cmd_analyze
|
||
client = _make_client()
|
||
with patch("builtins.print") as mock_print:
|
||
cmd_analyze(_args(path="/nonexistent/path/xyz"), client)
|
||
printed = "\n".join(str(c) for c in mock_print.call_args_list)
|
||
self.assertIn("不存在", printed)
|
||
|
||
|
||
class TestCmdAnalyzeNoAbapFiles(unittest.TestCase):
|
||
"""cmd_analyze 项目无 .abap 文件（空项目）。"""
|
||
|
||
def test_empty_project_no_crash(self):
|
||
from sapcli.commands.analyze import cmd_analyze
|
||
client = _make_client()
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with patch("builtins.print") as mock_print:
|
||
cmd_analyze(_args(path=td), client)
|
||
printed = "\n".join(str(c) for c in mock_print.call_args_list)
|
||
self.assertIn("未找到", printed)
|
||
|
||
|
||
class TestCmdAnalyzeNoDependencies(unittest.TestCase):
|
||
"""cmd_analyze 项目中 .abap 文件无外部依赖。"""
|
||
|
||
def test_no_z_deps_shows_no_dependencies(self):
|
||
from sapcli.commands.analyze import cmd_analyze
|
||
client = _make_client()
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
with open(os.path.join(report_dir, "zsimple.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT zsimple.\nWRITE: / 'hello'.\n")
|
||
|
||
with patch("builtins.print") as mock_print:
|
||
cmd_analyze(_args(path=td), client)
|
||
|
||
printed = "\n".join(str(c) for c in mock_print.call_args_list)
|
||
self.assertIn("无外部依赖", printed)
|
||
|
||
|
||
class TestCmdAnalyzeWithDependencies(unittest.TestCase):
|
||
"""cmd_analyze 正确检测 Z/Y 开头的外部依赖。"""
|
||
|
||
def test_detects_type_ref_to(self):
|
||
from sapcli.commands.analyze import cmd_analyze, _analyze_dependencies
|
||
source = "DATA: lo_obj TYPE REF TO ZCL_MY_CLASS."
|
||
deps = _analyze_dependencies(source)
|
||
self.assertIn("ZCL_MY_CLASS", deps)
|
||
|
||
def test_detects_call_function(self):
|
||
from sapcli.commands.analyze import _analyze_dependencies
|
||
source = "CALL FUNCTION 'Z_MY_FUNC'."
|
||
deps = _analyze_dependencies(source)
|
||
self.assertIn("Z_MY_FUNC", deps)
|
||
|
||
def test_detects_create_object(self):
|
||
from sapcli.commands.analyze import _analyze_dependencies
|
||
source = "CREATE OBJECT lo_obj TYPE ZCL_FACTORY."
|
||
deps = _analyze_dependencies(source)
|
||
self.assertIn("ZCL_FACTORY", deps)
|
||
|
||
def test_detects_type_reference(self):
|
||
from sapcli.commands.analyze import _analyze_dependencies
|
||
source = "DATA: lv_val TYPE ZCUSTOM_TYPE."
|
||
deps = _analyze_dependencies(source)
|
||
self.assertIn("ZCUSTOM_TYPE", deps)
|
||
|
||
def test_ignores_builtin_types(self):
|
||
from sapcli.commands.analyze import _analyze_dependencies
|
||
source = "DATA: lv_str TYPE STRING.\nDATA: lv_int TYPE INT4."
|
||
deps = _analyze_dependencies(source)
|
||
self.assertEqual(deps, [])
|
||
|
||
def test_ignores_short_names(self):
|
||
from sapcli.commands.analyze import _analyze_dependencies
|
||
source = "DATA: lv_val TYPE AB."
|
||
deps = _analyze_dependencies(source)
|
||
self.assertEqual(deps, [])
|
||
|
||
def test_detects_y_prefix(self):
|
||
from sapcli.commands.analyze import _analyze_dependencies
|
||
source = "DATA: lo TYPE REF TO YCL_SERVICE."
|
||
deps = _analyze_dependencies(source)
|
||
self.assertIn("YCL_SERVICE", deps)
|
||
|
||
def test_multiple_deps_deduplicated(self):
|
||
from sapcli.commands.analyze import _analyze_dependencies
|
||
source = (
|
||
"DATA: lo1 TYPE REF TO ZCL_A.\n"
|
||
"DATA: lo2 TYPE REF TO ZCL_A.\n"
|
||
"CALL FUNCTION 'ZCL_A'.\n"
|
||
)
|
||
deps = _analyze_dependencies(source)
|
||
# ZCL_A 只出现一次
|
||
self.assertEqual(deps.count("ZCL_A"), 1)
|
||
|
||
|
||
class TestCmdAnalyzeFileIntegration(unittest.TestCase):
|
||
"""cmd_analyze 完整文件分析集成测试。"""
|
||
|
||
def test_analyze_with_deps(self):
|
||
from sapcli.commands.analyze import cmd_analyze
|
||
client = _make_client()
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
with open(os.path.join(report_dir, "zrpt.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT zrpt.\nDATA: lo TYPE REF TO ZCL_HELPER.\n")
|
||
|
||
with patch("builtins.print") as mock_print:
|
||
cmd_analyze(_args(path=td), client)
|
||
|
||
printed = "\n".join(str(c) for c in mock_print.call_args_list)
|
||
self.assertIn("ZCL_HELPER", printed)
|
||
|
||
def test_analyze_multiple_files(self):
|
||
from sapcli.commands.analyze import cmd_analyze
|
||
client = _make_client()
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
# 文件 1: 有依赖
|
||
with open(os.path.join(report_dir, "za.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT za.\nDATA lo TYPE REF TO ZCL_DEP.\n")
|
||
# 文件 2: 无依赖
|
||
with open(os.path.join(report_dir, "zb.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT zb.\nWRITE: / 'hello'.\n")
|
||
|
||
with patch("builtins.print") as mock_print:
|
||
cmd_analyze(_args(path=td), client)
|
||
|
||
printed = "\n".join(str(c) for c in mock_print.call_args_list)
|
||
self.assertIn("2", printed) # 2 个文件
|
||
self.assertIn("ZCL_DEP", printed)
|
||
|
||
|
||
class TestCmdAnalyzeMissingFiles(unittest.TestCase):
|
||
"""cmd_analyze 处理缺少文件的对象。"""
|
||
|
||
def test_analyze_only_reads_existing_files(self):
|
||
"""cmd_analyze 通过 os.walk 扫描，只读存在的 .abap 文件，不会因 manifest 中
|
||
引用的缺失文件而崩溃。"""
|
||
from sapcli.commands.analyze import cmd_analyze
|
||
client = _make_client()
|
||
with tempfile.TemporaryDirectory() as td:
|
||
# 创建一个有文件的对象和一个空目录
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
with open(os.path.join(report_dir, "zexists.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT zexists.\n")
|
||
|
||
# 只分析到实际存在的文件
|
||
with patch("builtins.print") as mock_print:
|
||
cmd_analyze(_args(path=td), client)
|
||
|
||
printed = "\n".join(str(c) for c in mock_print.call_args_list)
|
||
self.assertIn("1", printed) # 1 个文件
|
||
|
||
|
||
class TestCmdAnalyzeEmptySource(unittest.TestCase):
|
||
"""_analyze_dependencies 输入空源码。"""
|
||
|
||
def test_empty_source_no_deps(self):
|
||
from sapcli.commands.analyze import _analyze_dependencies
|
||
deps = _analyze_dependencies("")
|
||
self.assertEqual(deps, [])
|
||
|
||
def test_whitespace_only_no_deps(self):
|
||
from sapcli.commands.analyze import _analyze_dependencies
|
||
deps = _analyze_dependencies(" \n \n ")
|
||
self.assertEqual(deps, [])
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# sorter.py — 环形依赖检测（补充）
|
||
# ═══════════════════════════════════════════
|
||
|
||
class TestTopologicalSortCyclic(unittest.TestCase):
|
||
"""topological_sort 检测循环依赖。"""
|
||
|
||
def test_simple_two_node_cycle(self):
|
||
from sapcli.sorter import topological_sort
|
||
from sapcli.exceptions import CyclicDependencyError
|
||
|
||
e1 = _make_entry(name="ZA", depends_on=["ZB"])
|
||
e2 = _make_entry(name="ZB", depends_on=["ZA"])
|
||
with self.assertRaises(CyclicDependencyError) as ctx:
|
||
topological_sort([e1, e2])
|
||
self.assertTrue(len(ctx.exception.cycle_members) > 0)
|
||
|
||
def test_three_node_cycle(self):
|
||
from sapcli.sorter import topological_sort
|
||
from sapcli.exceptions import CyclicDependencyError
|
||
|
||
e1 = _make_entry(name="ZA", depends_on=["ZC"])
|
||
e2 = _make_entry(name="ZB", depends_on=["ZA"])
|
||
e3 = _make_entry(name="ZC", depends_on=["ZB"])
|
||
with self.assertRaises(CyclicDependencyError):
|
||
topological_sort([e1, e2, e3])
|
||
|
||
def test_self_dependency(self):
|
||
from sapcli.sorter import topological_sort
|
||
from sapcli.exceptions import CyclicDependencyError
|
||
|
||
e1 = _make_entry(name="ZA", depends_on=["ZA"])
|
||
with self.assertRaises(CyclicDependencyError):
|
||
topological_sort([e1])
|
||
|
||
def test_no_deps_returns_all(self):
|
||
from sapcli.sorter import topological_sort
|
||
|
||
e1 = _make_entry(name="ZA")
|
||
e2 = _make_entry(name="ZB")
|
||
result = topological_sort([e1, e2])
|
||
names = [e.name for e in result]
|
||
self.assertEqual(set(names), {"ZA", "ZB"})
|
||
|
||
def test_external_dep_ignored(self):
|
||
"""depends_on 引用不在列表中的外部对象，不报错。"""
|
||
from sapcli.sorter import topological_sort
|
||
|
||
e1 = _make_entry(name="ZA", depends_on=["Z_EXTERNAL"])
|
||
result = topological_sort([e1])
|
||
self.assertEqual(len(result), 1)
|
||
|
||
def test_empty_list_returns_empty(self):
|
||
from sapcli.sorter import topological_sort
|
||
self.assertEqual(topological_sort([]), [])
|
||
|
||
|
||
# ═══════════════════════════════════════════
|
||
# batch.py — cmd_sync_all 成功路径
|
||
# ═══════════════════════════════════════════
|
||
|
||
class TestCmdSyncAllSuccess(unittest.TestCase):
|
||
"""cmd_sync_all 正常同步成功。"""
|
||
|
||
@patch("sapcli.commands.batch._sync_single")
|
||
def test_sync_updates_manifest(self, mock_sync):
|
||
from sapcli.commands.batch import cmd_sync_all
|
||
mock_sync.return_value = (True, None, "DEVK999")
|
||
client = _make_client()
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
report_dir = os.path.join(td, "reports")
|
||
os.makedirs(report_dir)
|
||
with open(os.path.join(report_dir, "ztest.abap"), "w", encoding="utf-8") as f:
|
||
f.write("REPORT ztest.")
|
||
|
||
e1 = _make_entry(status="inactive", last_sync_result="pending")
|
||
_write_manifest(td, [e1])
|
||
|
||
with patch("builtins.print"):
|
||
cmd_sync_all(_args(path=td), client)
|
||
|
||
# 验证 manifest 更新
|
||
with open(os.path.join(td, "manifest.json"), "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
self.assertEqual(data["objects"]["ZTEST"]["system_status"], "active")
|
||
self.assertEqual(data["objects"]["ZTEST"]["last_sync_result"], "success")
|
||
self.assertEqual(data["objects"]["ZTEST"]["corr_nr"], "DEVK999")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|