吴让宇
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 重写为单源开发流程。
899 lines
37 KiB
Python
899 lines
37 KiB
Python
"""散落模块（config / manifest / scanner / sorter / auth）单元测试。
|
||
|
||
运行: python tests/unit/test_modules.py
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import io
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "assets"))
|
||
|
||
from sapcli.config import SAPConfig, _find_config_file, list_profiles, load_config
|
||
from sapcli.exceptions import CyclicDependencyError, ConfigError
|
||
from sapcli.manifest import (
|
||
CURRENT_VERSION,
|
||
MANIFEST_FILENAME,
|
||
Manifest,
|
||
ManifestEntry,
|
||
init_manifest,
|
||
)
|
||
from sapcli.scanner import DIRECTORY_TYPE_MAP, ScannedObject, scan_project
|
||
from sapcli.sorter import TYPE_PRIORITY, topological_sort
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# config.py
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestSAPConfig(unittest.TestCase):
|
||
"""SAPConfig dataclass 创建与校验。"""
|
||
|
||
def test_create_valid(self):
|
||
cfg = SAPConfig(host="myhost", client="100", user="USER1", password="pass")
|
||
self.assertEqual(cfg.host, "myhost")
|
||
self.assertEqual(cfg.client, "100")
|
||
self.assertEqual(cfg.user, "USER1")
|
||
self.assertEqual(cfg.password, "pass")
|
||
|
||
def test_frozen(self):
|
||
cfg = SAPConfig(host="h", client="100", user="u", password="p")
|
||
with self.assertRaises(AttributeError):
|
||
cfg.host = "other" # type: ignore[misc]
|
||
|
||
def test_missing_host_raises(self):
|
||
with self.assertRaises(ConfigError):
|
||
SAPConfig(host="", client="100", user="u", password="p")
|
||
|
||
def test_missing_user_raises(self):
|
||
with self.assertRaises(ConfigError):
|
||
SAPConfig(host="h", client="100", user="", password="p")
|
||
|
||
def test_missing_password_raises(self):
|
||
with self.assertRaises(ConfigError):
|
||
SAPConfig(host="h", client="100", user="u", password="")
|
||
|
||
def test_missing_multiple_raises(self):
|
||
with self.assertRaises(ConfigError) as ctx:
|
||
SAPConfig(host="", client="100", user="", password="")
|
||
msg = str(ctx.exception)
|
||
self.assertIn("host", msg)
|
||
self.assertIn("user", msg)
|
||
|
||
|
||
class TestFindConfigFile(unittest.TestCase):
|
||
"""_find_config_file 搜索路径逻辑。"""
|
||
|
||
def test_returns_none_when_no_file(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
with patch("os.getcwd", return_value=tmp):
|
||
result = _find_config_file()
|
||
if result is not None:
|
||
self.assertTrue(os.path.isfile(result))
|
||
|
||
def test_finds_file_in_cwd(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
cfg_path = os.path.join(tmp, "config.ini")
|
||
with open(cfg_path, "w") as f:
|
||
f.write("[SAP]\nhost=example.com\n")
|
||
with patch("os.getcwd", return_value=tmp):
|
||
result = _find_config_file()
|
||
self.assertIsNotNone(result)
|
||
self.assertEqual(result, cfg_path)
|
||
|
||
|
||
def _write_tmp_ini(content: str) -> str:
|
||
"""写临时 .ini 文件，返回路径（文件已关闭，可在 Windows 上删除）。"""
|
||
fd, path = tempfile.mkstemp(suffix=".ini")
|
||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
return path
|
||
|
||
|
||
class TestListProfiles(unittest.TestCase):
|
||
"""list_profiles 从 config.ini 读取 sections。"""
|
||
|
||
def test_reads_sections(self):
|
||
path = _write_tmp_ini("[SAP]\nhost=h1\n[DEV]\nhost=h2\n[PRD]\nhost=h3\n")
|
||
try:
|
||
profiles = list_profiles(config_path=path)
|
||
self.assertIn("SAP", profiles)
|
||
self.assertIn("DEV", profiles)
|
||
self.assertIn("PRD", profiles)
|
||
self.assertEqual(len(profiles), 3)
|
||
finally:
|
||
os.unlink(path)
|
||
|
||
def test_empty_file_returns_empty(self):
|
||
path = _write_tmp_ini("")
|
||
try:
|
||
profiles = list_profiles(config_path=path)
|
||
self.assertEqual(profiles, [])
|
||
finally:
|
||
os.unlink(path)
|
||
|
||
def test_no_config_path_no_file(self):
|
||
with patch("sapcli.config._find_config_file", return_value=None):
|
||
profiles = list_profiles(config_path=None)
|
||
self.assertEqual(profiles, [])
|
||
|
||
|
||
class TestLoadConfig(unittest.TestCase):
|
||
"""load_config 从文件和环境变量读取。"""
|
||
|
||
@staticmethod
|
||
def _write_config(path: str, section: str = "SAP", **kwargs: str) -> None:
|
||
import configparser
|
||
cfg = configparser.ConfigParser()
|
||
cfg[section] = kwargs
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
cfg.write(f)
|
||
|
||
def _clean_env(self):
|
||
return {k: v for k, v in os.environ.items()
|
||
if k not in ("SAP_HOST", "SAP_CLIENT", "SAP_USER", "SAP_PASSWORD")}
|
||
|
||
def test_load_from_file(self):
|
||
fd, path = tempfile.mkstemp(suffix=".ini")
|
||
os.close(fd)
|
||
try:
|
||
self._write_config(path, host="myhost.com", client="200",
|
||
user="MYUSER", password="secret")
|
||
with patch.dict(os.environ, self._clean_env(), clear=True):
|
||
cfg, loaded_from = load_config(config_path=path)
|
||
self.assertEqual(cfg.host, "myhost.com")
|
||
self.assertEqual(cfg.client, "200")
|
||
self.assertEqual(cfg.user, "MYUSER")
|
||
self.assertIsNotNone(loaded_from)
|
||
finally:
|
||
os.unlink(path)
|
||
|
||
def test_env_overrides_file(self):
|
||
fd, path = tempfile.mkstemp(suffix=".ini")
|
||
os.close(fd)
|
||
try:
|
||
self._write_config(path, host="file-host", client="100",
|
||
user="file-user", password="file-pass")
|
||
env = self._clean_env()
|
||
env["SAP_HOST"] = "env-host"
|
||
env["SAP_USER"] = "env-user"
|
||
env["SAP_PASSWORD"] = "env-pass"
|
||
with patch.dict(os.environ, env, clear=True):
|
||
cfg, _ = load_config(config_path=path)
|
||
self.assertEqual(cfg.host, "env-host")
|
||
self.assertEqual(cfg.user, "env-user")
|
||
finally:
|
||
os.unlink(path)
|
||
|
||
def test_missing_host_raises_config_error(self):
|
||
fd, path = tempfile.mkstemp(suffix=".ini")
|
||
os.close(fd)
|
||
try:
|
||
self._write_config(path, client="100", user="u", password="p")
|
||
with patch.dict(os.environ, self._clean_env(), clear=True):
|
||
with self.assertRaises(ConfigError):
|
||
load_config(config_path=path)
|
||
finally:
|
||
os.unlink(path)
|
||
|
||
def test_profile_parameter(self):
|
||
fd, path = tempfile.mkstemp(suffix=".ini")
|
||
os.close(fd)
|
||
try:
|
||
self._write_config(path, section="DEV",
|
||
host="dev-host", client="400",
|
||
user="DEVUSER", password="devpass")
|
||
with patch.dict(os.environ, self._clean_env(), clear=True):
|
||
cfg, _ = load_config(config_path=path, profile="DEV")
|
||
self.assertEqual(cfg.host, "dev-host")
|
||
self.assertEqual(cfg.client, "400")
|
||
finally:
|
||
os.unlink(path)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# manifest.py
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestManifestEntry(unittest.TestCase):
|
||
"""ManifestEntry 序列化 / 反序列化。"""
|
||
|
||
def test_to_dict_defaults(self):
|
||
entry = ManifestEntry(name="ZFOO", type="report", file="reports/zfoo.abap")
|
||
d = entry.to_dict()
|
||
self.assertEqual(d["type"], "report")
|
||
self.assertEqual(d["file"], "reports/zfoo.abap")
|
||
self.assertEqual(d["system_status"], "not_exists")
|
||
self.assertIsNone(d["corr_nr"])
|
||
self.assertEqual(d["depends_on"], [])
|
||
self.assertIsNone(d["last_sync"])
|
||
self.assertEqual(d["last_sync_result"], "pending")
|
||
|
||
def test_to_dict_full(self):
|
||
entry = ManifestEntry(
|
||
name="ZBAR",
|
||
type="class",
|
||
file="classes/zbar.abap",
|
||
system_status="active",
|
||
corr_nr="NR001",
|
||
depends_on=["ZFOO"],
|
||
last_sync="2026-01-01T00:00:00",
|
||
last_sync_result="success",
|
||
)
|
||
d = entry.to_dict()
|
||
self.assertEqual(d["system_status"], "active")
|
||
self.assertEqual(d["corr_nr"], "NR001")
|
||
self.assertEqual(d["depends_on"], ["ZFOO"])
|
||
self.assertEqual(d["last_sync"], "2026-01-01T00:00:00")
|
||
self.assertEqual(d["last_sync_result"], "success")
|
||
|
||
def test_from_dict_roundtrip(self):
|
||
original = ManifestEntry(
|
||
name="ZRT",
|
||
type="table",
|
||
file="tables/zrt.abap",
|
||
system_status="inactive",
|
||
corr_nr="C002",
|
||
depends_on=["ZA", "ZB"],
|
||
last_sync="2026-06-01T12:00:00",
|
||
last_sync_result="failed",
|
||
)
|
||
d = original.to_dict()
|
||
restored = ManifestEntry.from_dict("ZRT", d)
|
||
self.assertEqual(restored.name, original.name)
|
||
self.assertEqual(restored.type, original.type)
|
||
self.assertEqual(restored.file, original.file)
|
||
self.assertEqual(restored.system_status, original.system_status)
|
||
self.assertEqual(restored.corr_nr, original.corr_nr)
|
||
self.assertEqual(restored.depends_on, original.depends_on)
|
||
self.assertEqual(restored.last_sync, original.last_sync)
|
||
self.assertEqual(restored.last_sync_result, original.last_sync_result)
|
||
|
||
def test_from_dict_defaults(self):
|
||
entry = ManifestEntry.from_dict("ZNEW", {"type": "domain", "file": "domains/znew.abap"})
|
||
self.assertEqual(entry.name, "ZNEW")
|
||
self.assertEqual(entry.system_status, "not_exists")
|
||
self.assertIsNone(entry.corr_nr)
|
||
self.assertEqual(entry.depends_on, [])
|
||
self.assertIsNone(entry.last_sync)
|
||
self.assertEqual(entry.last_sync_result, "pending")
|
||
|
||
|
||
class TestManifest(unittest.TestCase):
|
||
"""Manifest load/save 往返和 CRUD。"""
|
||
|
||
def test_save_and_load_roundtrip(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
manifest = Manifest(
|
||
version=CURRENT_VERSION,
|
||
last_init="2026-01-01T00:00:00",
|
||
last_refresh=None,
|
||
objects={
|
||
"ZFOO": ManifestEntry(
|
||
name="ZFOO", type="report", file="reports/zfoo.abap",
|
||
system_status="active", last_sync_result="success",
|
||
),
|
||
"ZBAR": ManifestEntry(
|
||
name="ZBAR", type="class", file="classes/zbar.abap",
|
||
depends_on=["ZFOO"],
|
||
),
|
||
},
|
||
_project_path=tmp,
|
||
)
|
||
manifest.save()
|
||
|
||
filepath = os.path.join(tmp, MANIFEST_FILENAME)
|
||
self.assertTrue(os.path.isfile(filepath))
|
||
|
||
loaded = Manifest.load(tmp)
|
||
self.assertEqual(loaded.version, CURRENT_VERSION)
|
||
self.assertEqual(len(loaded.objects), 2)
|
||
self.assertIn("ZFOO", loaded.objects)
|
||
self.assertIn("ZBAR", loaded.objects)
|
||
self.assertEqual(loaded.objects["ZFOO"].type, "report")
|
||
self.assertEqual(loaded.objects["ZBAR"].depends_on, ["ZFOO"])
|
||
|
||
def test_load_missing_file_raises(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
with self.assertRaises(FileNotFoundError):
|
||
Manifest.load(tmp)
|
||
|
||
def test_upsert_and_remove(self):
|
||
m = Manifest()
|
||
entry = ManifestEntry(name="Z1", type="report", file="reports/z1.abap")
|
||
m.upsert(entry)
|
||
self.assertIn("Z1", m.objects)
|
||
self.assertEqual(m.get("Z1").type, "report")
|
||
|
||
entry2 = ManifestEntry(name="Z1", type="class", file="classes/z1.abap")
|
||
m.upsert(entry2)
|
||
self.assertEqual(m.get("Z1").type, "class")
|
||
|
||
self.assertTrue(m.remove("Z1"))
|
||
self.assertIsNone(m.get("Z1"))
|
||
self.assertFalse(m.remove("Z1"))
|
||
|
||
def test_pending_objects(self):
|
||
m = Manifest()
|
||
m.upsert(ManifestEntry(name="A", type="report", file="r/a.abap",
|
||
system_status="active", last_sync_result="success"))
|
||
m.upsert(ManifestEntry(name="B", type="class", file="c/b.abap",
|
||
system_status="not_exists", last_sync_result="pending"))
|
||
m.upsert(ManifestEntry(name="C", type="table", file="t/c.abap",
|
||
system_status="active", last_sync_result="failed"))
|
||
pending = m.pending_objects()
|
||
names = [e.name for e in pending]
|
||
self.assertIn("B", names)
|
||
self.assertIn("C", names)
|
||
self.assertNotIn("A", names)
|
||
|
||
def test_active_up_to_date_objects(self):
|
||
m = Manifest()
|
||
m.upsert(ManifestEntry(name="A", type="report", file="r/a.abap",
|
||
system_status="active", last_sync_result="success"))
|
||
m.upsert(ManifestEntry(name="B", type="class", file="c/b.abap",
|
||
system_status="not_exists", last_sync_result="pending"))
|
||
active = m.active_up_to_date_objects()
|
||
self.assertEqual(len(active), 1)
|
||
self.assertEqual(active[0].name, "A")
|
||
|
||
def test_file_path(self):
|
||
m = Manifest(_project_path="/project")
|
||
entry = ManifestEntry(name="Z1", type="report", file="reports/z1.abap")
|
||
fp = m.file_path(entry)
|
||
expected = os.path.join("/project", "reports", "z1.abap")
|
||
self.assertEqual(fp, expected)
|
||
|
||
|
||
class TestInitManifest(unittest.TestCase):
|
||
"""init_manifest 创建空清单。"""
|
||
|
||
def test_creates_manifest_with_init_time(self):
|
||
m = init_manifest("/tmp/fake")
|
||
self.assertEqual(m.version, CURRENT_VERSION)
|
||
self.assertIsNotNone(m.last_init)
|
||
self.assertEqual(len(m.objects), 0)
|
||
self.assertEqual(m._project_path, "/tmp/fake")
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# scanner.py
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestScanProject(unittest.TestCase):
|
||
"""scan_project 目录扫描。"""
|
||
|
||
def _make_project(self, tmp: str) -> None:
|
||
"""创建模拟项目目录结构。"""
|
||
# reports/zrpt001.abap
|
||
os.makedirs(os.path.join(tmp, "reports"))
|
||
with open(os.path.join(tmp, "reports", "zrpt001.abap"), "w") as f:
|
||
f.write("REPORT zrpt001.\n")
|
||
# reports/.hidden.abap — 应被跳过
|
||
with open(os.path.join(tmp, "reports", ".hidden.abap"), "w") as f:
|
||
f.write("REPORT hidden.\n")
|
||
# reports/notes.txt — 非 .abap 应被跳过
|
||
with open(os.path.join(tmp, "reports", "notes.txt"), "w") as f:
|
||
f.write("notes\n")
|
||
|
||
# classes/zcl_demo.abap
|
||
os.makedirs(os.path.join(tmp, "classes"))
|
||
with open(os.path.join(tmp, "classes", "zcl_demo.abap"), "w") as f:
|
||
f.write("CLASS zcl_demo DEFINITION.\n")
|
||
|
||
# functions/zgroup/zfunc1.abap
|
||
os.makedirs(os.path.join(tmp, "functions", "zgroup"))
|
||
with open(os.path.join(tmp, "functions", "zgroup", "zfunc1.abap"), "w") as f:
|
||
f.write("FUNCTION zfunc1.\n")
|
||
# functions/zgroup/zfunc2.abap
|
||
with open(os.path.join(tmp, "functions", "zgroup", "zfunc2.abap"), "w") as f:
|
||
f.write("FUNCTION zfunc2.\n")
|
||
|
||
# functions/.hidden_grp/func.abap — 隐藏目录应被跳过
|
||
os.makedirs(os.path.join(tmp, "functions", ".hidden_grp"))
|
||
with open(os.path.join(tmp, "functions", ".hidden_grp", "func.abap"), "w") as f:
|
||
f.write("FUNCTION func.\n")
|
||
|
||
# domains/zdom_test.abap
|
||
os.makedirs(os.path.join(tmp, "domains"))
|
||
with open(os.path.join(tmp, "domains", "zdom_test.abap"), "w") as f:
|
||
f.write("DOMAIN zdom_test.\n")
|
||
|
||
def test_scan_finds_all_objects(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
self._make_project(tmp)
|
||
results = scan_project(tmp)
|
||
names = {r.name for r in results}
|
||
|
||
self.assertIn("ZRPT001", names)
|
||
self.assertNotIn("HIDDEN", names)
|
||
self.assertIn("ZCL_DEMO", names)
|
||
self.assertIn("ZGROUP/ZFUNC1", names)
|
||
self.assertIn("ZGROUP/ZFUNC2", names)
|
||
self.assertNotIn("HIDDEN_GRP/FUNC", names)
|
||
self.assertIn("ZDOM_TEST", names)
|
||
|
||
def test_scan_types(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
self._make_project(tmp)
|
||
results = scan_project(tmp)
|
||
type_map = {r.name: r.type for r in results}
|
||
|
||
self.assertEqual(type_map.get("ZRPT001"), "report")
|
||
self.assertEqual(type_map.get("ZCL_DEMO"), "class")
|
||
self.assertEqual(type_map.get("ZGROUP/ZFUNC1"), "function")
|
||
self.assertEqual(type_map.get("ZGROUP/ZFUNC2"), "function")
|
||
self.assertEqual(type_map.get("ZDOM_TEST"), "domain")
|
||
|
||
def test_scan_file_paths(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
self._make_project(tmp)
|
||
results = scan_project(tmp)
|
||
file_map = {r.name: r.file for r in results}
|
||
|
||
self.assertEqual(file_map["ZRPT001"], "reports/zrpt001.abap")
|
||
self.assertEqual(file_map["ZCL_DEMO"], "classes/zcl_demo.abap")
|
||
self.assertEqual(file_map["ZGROUP/ZFUNC1"], "functions/zgroup/zfunc1.abap")
|
||
self.assertEqual(file_map["ZDOM_TEST"], "domains/zdom_test.abap")
|
||
|
||
def test_scan_empty_directory(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
results = scan_project(tmp)
|
||
self.assertEqual(results, [])
|
||
|
||
def test_scanned_object_dataclass(self):
|
||
obj = ScannedObject(name="ZTEST", type="report", file="reports/ztest.abap")
|
||
self.assertEqual(obj.name, "ZTEST")
|
||
self.assertEqual(obj.type, "report")
|
||
self.assertEqual(obj.file, "reports/ztest.abap")
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# sorter.py
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestTopologicalSort(unittest.TestCase):
|
||
"""topological_sort 按类型优先级和依赖排序。"""
|
||
|
||
def _entry(self, name: str, obj_type: str, depends_on: list[str] | None = None) -> ManifestEntry:
|
||
return ManifestEntry(
|
||
name=name, type=obj_type, file=f"{obj_type}s/{name.lower()}.abap",
|
||
depends_on=depends_on or [],
|
||
)
|
||
|
||
def test_empty_list(self):
|
||
self.assertEqual(topological_sort([]), [])
|
||
|
||
def test_sorts_by_type_priority(self):
|
||
"""无依赖时，按 TYPE_PRIORITY 数值从小到大排序。"""
|
||
objs = [
|
||
self._entry("ZCL1", "class"),
|
||
self._entry("ZDOM1", "domain"),
|
||
self._entry("ZRPT1", "report"),
|
||
self._entry("ZTAB1", "table"),
|
||
]
|
||
result = topological_sort(objs)
|
||
names = [o.name for o in result]
|
||
self.assertEqual(names, ["ZDOM1", "ZTAB1", "ZCL1", "ZRPT1"])
|
||
|
||
def test_respects_depends_on(self):
|
||
"""有依赖时，被依赖的对象排在前面。"""
|
||
objs = [
|
||
self._entry("ZRPT1", "report", depends_on=["ZCL1"]),
|
||
self._entry("ZCL1", "class", depends_on=["ZTAB1"]),
|
||
self._entry("ZTAB1", "table"),
|
||
]
|
||
result = topological_sort(objs)
|
||
names = [o.name for o in result]
|
||
idx_tab = names.index("ZTAB1")
|
||
idx_cl = names.index("ZCL1")
|
||
idx_rpt = names.index("ZRPT1")
|
||
self.assertLess(idx_tab, idx_cl)
|
||
self.assertLess(idx_cl, idx_rpt)
|
||
|
||
def test_cyclic_dependency_raises(self):
|
||
"""循环依赖应抛出 CyclicDependencyError。"""
|
||
objs = [
|
||
self._entry("A", "report", depends_on=["B"]),
|
||
self._entry("B", "class", depends_on=["A"]),
|
||
]
|
||
with self.assertRaises(CyclicDependencyError):
|
||
topological_sort(objs)
|
||
|
||
def test_external_dependency_ignored(self):
|
||
"""depends_on 引用不在列表中的对象时，忽略该依赖。"""
|
||
objs = [
|
||
self._entry("ZRPT1", "report", depends_on=["EXTERNAL"]),
|
||
]
|
||
result = topological_sort(objs)
|
||
self.assertEqual(len(result), 1)
|
||
self.assertEqual(result[0].name, "ZRPT1")
|
||
|
||
def test_unknown_type_gets_default_priority(self):
|
||
"""未注册类型应获得默认优先级 (90)。"""
|
||
objs = [
|
||
self._entry("ZUNKNOWN", "weird_type"),
|
||
self._entry("ZRPT1", "report"),
|
||
]
|
||
result = topological_sort(objs)
|
||
names = [o.name for o in result]
|
||
self.assertEqual(names, ["ZRPT1", "ZUNKNOWN"])
|
||
|
||
def test_diamond_dependency(self):
|
||
"""菱形依赖: A → B, A → C, B → D, C → D。D 最先。"""
|
||
objs = [
|
||
self._entry("A", "report", depends_on=["B", "C"]),
|
||
self._entry("B", "class", depends_on=["D"]),
|
||
self._entry("C", "function", depends_on=["D"]),
|
||
self._entry("D", "domain"),
|
||
]
|
||
result = topological_sort(objs)
|
||
names = [o.name for o in result]
|
||
idx_d = names.index("D")
|
||
idx_b = names.index("B")
|
||
idx_c = names.index("C")
|
||
idx_a = names.index("A")
|
||
self.assertLess(idx_d, idx_b)
|
||
self.assertLess(idx_d, idx_c)
|
||
self.assertLess(idx_b, idx_a)
|
||
self.assertLess(idx_c, idx_a)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
# auth.py
|
||
# ═══════════════════════════════════════════════════════════════════
|
||
|
||
class TestServiceName(unittest.TestCase):
|
||
"""_service_name 格式。"""
|
||
|
||
def test_format(self):
|
||
from sapcli.auth import _service_name
|
||
self.assertEqual(_service_name("myhost.com", "100"), "sap-cli:myhost.com:100")
|
||
|
||
def test_format_empty_client(self):
|
||
from sapcli.auth import _service_name
|
||
self.assertEqual(_service_name("host", ""), "sap-cli:host:")
|
||
|
||
|
||
class _AuthTestBase(unittest.TestCase):
|
||
"""auth 测试基类，mock keyring 模块。"""
|
||
|
||
def _patch_keyring(self):
|
||
"""将 sapcli.auth / sapcli.password 的 _keyring / _KEYRING_AVAILABLE 替换为 mock。"""
|
||
import sapcli.auth as auth_mod
|
||
import sapcli.password as pwd_mod
|
||
|
||
self.mock_kr = MagicMock()
|
||
# mock keyring module with errors attribute
|
||
self.mock_kr_mod = MagicMock()
|
||
self.mock_kr_mod.errors.PasswordDeleteError = type("PasswordDeleteError", (Exception,), {})
|
||
|
||
# 保存原始值以便恢复
|
||
self._orig_keyring = getattr(auth_mod, "_keyring", None)
|
||
self._orig_available = getattr(auth_mod, "_KEYRING_AVAILABLE", None)
|
||
self._orig_kr_mod = getattr(auth_mod, "_keyring_mod", None)
|
||
self._had_kr_mod = hasattr(auth_mod, "_keyring_mod")
|
||
|
||
self._orig_pwd_keyring = getattr(pwd_mod, "_keyring", None)
|
||
self._orig_pwd_available = getattr(pwd_mod, "_KEYRING_AVAILABLE", None)
|
||
self._orig_pwd_kr_mod = getattr(pwd_mod, "_keyring_mod", None)
|
||
self._had_pwd_kr_mod = hasattr(pwd_mod, "_keyring_mod")
|
||
|
||
# auth 模块
|
||
auth_mod._keyring = self.mock_kr
|
||
auth_mod._KEYRING_AVAILABLE = True
|
||
auth_mod._keyring_mod = self.mock_kr_mod
|
||
|
||
# password 模块（resolve_password / get_password 实际所在）
|
||
pwd_mod._keyring = self.mock_kr
|
||
pwd_mod._KEYRING_AVAILABLE = True
|
||
pwd_mod._keyring_mod = self.mock_kr_mod
|
||
|
||
self.addCleanup(self._restore_keyring)
|
||
|
||
def _restore_keyring(self):
|
||
import sapcli.auth as auth_mod
|
||
import sapcli.password as pwd_mod
|
||
auth_mod._keyring = self._orig_keyring
|
||
auth_mod._KEYRING_AVAILABLE = self._orig_available
|
||
if self._had_kr_mod:
|
||
auth_mod._keyring_mod = self._orig_kr_mod
|
||
elif hasattr(auth_mod, "_keyring_mod"):
|
||
del auth_mod._keyring_mod
|
||
|
||
pwd_mod._keyring = self._orig_pwd_keyring
|
||
pwd_mod._KEYRING_AVAILABLE = self._orig_pwd_available
|
||
if self._had_pwd_kr_mod:
|
||
pwd_mod._keyring_mod = self._orig_pwd_kr_mod
|
||
elif hasattr(pwd_mod, "_keyring_mod"):
|
||
del pwd_mod._keyring_mod
|
||
|
||
|
||
class TestGetPassword(_AuthTestBase):
|
||
"""get_password 从 keyring 读取密码。"""
|
||
|
||
def setUp(self):
|
||
self._patch_keyring()
|
||
|
||
def test_returns_password(self):
|
||
from sapcli.auth import get_password
|
||
self.mock_kr.get_password.return_value = "secret123"
|
||
result = get_password("host", "100", "user1")
|
||
self.assertEqual(result, "secret123")
|
||
self.mock_kr.get_password.assert_called_once_with("sap-cli:host:100", "user1")
|
||
|
||
def test_returns_none_on_exception(self):
|
||
from sapcli.auth import get_password
|
||
self.mock_kr.get_password.side_effect = Exception("boom")
|
||
result = get_password("host", "100", "user1")
|
||
self.assertIsNone(result)
|
||
|
||
def test_returns_none_when_unavailable(self):
|
||
from sapcli.auth import get_password
|
||
with patch("sapcli.password._KEYRING_AVAILABLE", False):
|
||
result = get_password("host", "100", "user1")
|
||
self.assertIsNone(result)
|
||
|
||
|
||
class TestSetPassword(_AuthTestBase):
|
||
"""set_password 存储密码到 keyring。"""
|
||
|
||
def setUp(self):
|
||
self._patch_keyring()
|
||
|
||
def test_success(self):
|
||
from sapcli.auth import set_password
|
||
result = set_password("host", "100", "user1", "pass")
|
||
self.assertTrue(result)
|
||
self.mock_kr.set_password.assert_called_once_with("sap-cli:host:100", "user1", "pass")
|
||
|
||
def test_failure_returns_false(self):
|
||
from sapcli.auth import set_password
|
||
self.mock_kr.set_password.side_effect = Exception("fail")
|
||
result = set_password("host", "100", "user1", "pass")
|
||
self.assertFalse(result)
|
||
|
||
def test_unavailable_returns_false(self):
|
||
from sapcli.auth import set_password
|
||
with patch("sapcli.auth._KEYRING_AVAILABLE", False):
|
||
result = set_password("host", "100", "user1", "pass")
|
||
self.assertFalse(result)
|
||
|
||
|
||
class TestDeletePassword(_AuthTestBase):
|
||
"""delete_password 从 keyring 删除密码。"""
|
||
|
||
def setUp(self):
|
||
self._patch_keyring()
|
||
|
||
def test_success(self):
|
||
from sapcli.auth import delete_password
|
||
result = delete_password("host", "100", "user1")
|
||
self.assertTrue(result)
|
||
self.mock_kr.delete_password.assert_called_once_with("sap-cli:host:100", "user1")
|
||
|
||
def test_failure_returns_false(self):
|
||
from sapcli.auth import delete_password
|
||
self.mock_kr.delete_password.side_effect = Exception("fail")
|
||
result = delete_password("host", "100", "user1")
|
||
self.assertFalse(result)
|
||
|
||
def test_unavailable_returns_false(self):
|
||
from sapcli.auth import delete_password
|
||
with patch("sapcli.auth._KEYRING_AVAILABLE", False):
|
||
result = delete_password("host", "100", "user1")
|
||
self.assertFalse(result)
|
||
|
||
|
||
class TestResolvePassword(_AuthTestBase):
|
||
"""resolve_password 按优先级解析密码。"""
|
||
|
||
def setUp(self):
|
||
self._patch_keyring()
|
||
|
||
def test_env_password_takes_priority(self):
|
||
from sapcli.auth import resolve_password
|
||
self.mock_kr.get_password.return_value = "kr_pass"
|
||
result = resolve_password("h", "100", "u", config_password="cfg_pass", env_password="env_pass")
|
||
self.assertEqual(result, "env_pass")
|
||
|
||
def test_keyring_over_config(self):
|
||
from sapcli.auth import resolve_password
|
||
self.mock_kr.get_password.return_value = "kr_pass"
|
||
result = resolve_password("h", "100", "u", config_password="cfg_pass", env_password="")
|
||
self.assertEqual(result, "kr_pass")
|
||
|
||
def test_config_as_fallback(self):
|
||
from sapcli.auth import resolve_password
|
||
self.mock_kr.get_password.return_value = None
|
||
result = resolve_password("h", "100", "u", config_password="cfg_pass", env_password="")
|
||
self.assertEqual(result, "cfg_pass")
|
||
|
||
def test_all_empty(self):
|
||
from sapcli.auth import resolve_password
|
||
self.mock_kr.get_password.return_value = None
|
||
result = resolve_password("h", "100", "u", config_password="", env_password="")
|
||
self.assertEqual(result, "")
|
||
|
||
|
||
class TestCmdAuthStatus(unittest.TestCase):
|
||
"""cmd_auth_status 输出。"""
|
||
|
||
def test_prints_keyring_available(self):
|
||
from sapcli.auth import cmd_auth_status
|
||
mock_backend = MagicMock()
|
||
mock_backend.__class__ = type("TestBackend", (), {})
|
||
with patch("sapcli.auth._KEYRING_AVAILABLE", True), \
|
||
patch("sapcli.auth._keyring") as mock_kr:
|
||
mock_kr.get_keyring.return_value = mock_backend
|
||
with patch("sys.stdout", new_callable=io.StringIO) as out:
|
||
cmd_auth_status(argparse.Namespace())
|
||
output = out.getvalue()
|
||
self.assertIn("keyring", output)
|
||
self.assertIn("可用", output)
|
||
|
||
def test_prints_keyring_unavailable(self):
|
||
from sapcli.auth import cmd_auth_status
|
||
with patch("sapcli.auth._KEYRING_AVAILABLE", False):
|
||
with patch("sys.stdout", new_callable=io.StringIO) as out:
|
||
cmd_auth_status(argparse.Namespace())
|
||
output = out.getvalue()
|
||
self.assertIn("keyring", output)
|
||
self.assertIn("不可用", output)
|
||
|
||
|
||
class TestCmdAuthLogin(unittest.TestCase):
|
||
"""cmd_auth_login 交互式登录。"""
|
||
|
||
def test_login_stores_password(self):
|
||
from sapcli.auth import cmd_auth_login
|
||
|
||
fake_cfg = SAPConfig(host="h", client="100", user="u", password="p")
|
||
args = argparse.Namespace(config=None)
|
||
|
||
with patch("sapcli.config.load_config", return_value=(fake_cfg, "/fake")), \
|
||
patch("sapcli.auth.getpass") as mock_gp_mod, \
|
||
patch("sapcli.auth.set_password", return_value=True) as mock_sp, \
|
||
patch("sys.stdout", new_callable=io.StringIO):
|
||
|
||
mock_gp_mod.getpass.return_value = "mypassword"
|
||
cmd_auth_login(args)
|
||
|
||
mock_sp.assert_called_once_with("h", "100", "u", "mypassword")
|
||
|
||
def test_login_empty_password_aborts(self):
|
||
from sapcli.auth import cmd_auth_login
|
||
|
||
fake_cfg = SAPConfig(host="h", client="100", user="u", password="p")
|
||
|
||
with patch("sapcli.config.load_config", return_value=(fake_cfg, "/fake")), \
|
||
patch("sapcli.auth.getpass.getpass", return_value=""), \
|
||
patch("sapcli.auth.set_password") as mock_sp, \
|
||
patch("sys.stdout", new_callable=io.StringIO) as out:
|
||
|
||
cmd_auth_login(argparse.Namespace(config=None))
|
||
# 不应调用 set_password
|
||
mock_sp.assert_not_called()
|
||
self.assertIn("不能为空", out.getvalue())
|
||
|
||
|
||
class TestCmdAuthLogout(unittest.TestCase):
|
||
"""cmd_auth_logout 删除密码。"""
|
||
|
||
def test_logout_deletes_password(self):
|
||
from sapcli.auth import cmd_auth_logout
|
||
|
||
fake_cfg = SAPConfig(host="h", client="100", user="u", password="p")
|
||
|
||
with patch("sapcli.config.load_config", return_value=(fake_cfg, "/fake")), \
|
||
patch("sapcli.auth.delete_password", return_value=True) as mock_dp, \
|
||
patch("sys.stdout", new_callable=io.StringIO) as out:
|
||
|
||
cmd_auth_logout(argparse.Namespace(config=None))
|
||
|
||
mock_dp.assert_called_once_with("h", "100", "u")
|
||
self.assertIn("删除", out.getvalue())
|
||
|
||
class TestScanProjectThreeLayer(unittest.TestCase):
|
||
"""scan_project 三层布局（src/<开发包>/<对象类型>/）+ 单复数目录名。"""
|
||
|
||
def _make_project(self, tmp: str) -> None:
|
||
"""src/TMP/<类型>/ 三层布局。"""
|
||
for sub in (
|
||
"src/TMP/report", "src/TMP/class", "src/TMP/interface",
|
||
"src/TMP/domain", "src/TMP/structure", "src/TMP/tabletype",
|
||
):
|
||
os.makedirs(os.path.join(tmp, *sub.split("/")))
|
||
|
||
def put(rel: str, body: str) -> None:
|
||
with open(os.path.join(tmp, *rel.split("/")), "w") as f:
|
||
f.write(body)
|
||
|
||
put("src/TMP/report/zrpt001.abap", "REPORT zrpt001.\n")
|
||
put("src/TMP/report/notes.txt", "skip me\n") # 非 .abap 跳过
|
||
put("src/TMP/class/zcl_demo.abap", "CLASS zcl_demo DEFINITION.\n")
|
||
put("src/TMP/interface/zif_demo.abap", "INTERFACE zif_demo.\n")
|
||
put("src/TMP/domain/zdom_demo.abap", "DOMAIN zdom_demo.\n")
|
||
put("src/TMP/structure/zst_demo.abap", "STRUCTURE zst_demo.\n")
|
||
put("src/TMP/tabletype/ztt_demo.abap", "TABLETYPE ztt_demo.\n")
|
||
|
||
def _make_funcs(self, tmp: str) -> None:
|
||
"""函数模块两种命名。"""
|
||
os.makedirs(os.path.join(tmp, "src", "TMP", "function", "zfg_one"))
|
||
with open(os.path.join(tmp, "src", "TMP", "function", "zfg_one", "zfm_a.abap"), "w") as f:
|
||
f.write("FUNCTION zfm_a.\n")
|
||
with open(os.path.join(tmp, "src", "TMP", "function", "zfg_two.fugr.zfm_b.abap"), "w") as f:
|
||
f.write("FUNCTION zfm_b.\n")
|
||
|
||
def test_three_layer_finds_objects(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
self._make_project(tmp)
|
||
results = scan_project(tmp)
|
||
by_name = {r.name: r for r in results}
|
||
|
||
self.assertIn("ZRPT001", by_name)
|
||
self.assertNotIn("NOTES", by_name)
|
||
self.assertIn("ZCL_DEMO", by_name)
|
||
self.assertIn("ZIF_DEMO", by_name)
|
||
self.assertIn("ZDOM_DEMO", by_name)
|
||
self.assertIn("ZST_DEMO", by_name)
|
||
self.assertIn("ZTT_DEMO", by_name)
|
||
|
||
# 类型识别正确（按目录名，而非文件名）
|
||
self.assertEqual(by_name["ZCL_DEMO"].type, "class")
|
||
self.assertEqual(by_name["ZST_DEMO"].type, "structure")
|
||
self.assertEqual(by_name["ZTT_DEMO"].type, "tabletype")
|
||
# 相对项目根路径保留三层
|
||
self.assertEqual(by_name["ZCL_DEMO"].file, "src/TMP/class/zcl_demo.abap")
|
||
|
||
def test_function_two_naming_styles(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
self._make_funcs(tmp)
|
||
names = {r.name for r in scan_project(tmp)}
|
||
self.assertIn("ZFG_ONE/ZFM_A", names) # 子目录式
|
||
self.assertIn("ZFG_TWO/ZFM_B", names) # ADT 式 <fugr>.fugr.<fm>
|
||
|
||
def test_plural_and_case_insensitive(self):
|
||
"""classes/ 与 class/ 等价，目录名大小写不敏感。"""
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
os.makedirs(os.path.join(tmp, "src", "TMP", "CLASSES"))
|
||
with open(os.path.join(tmp, "src", "TMP", "CLASSES", "zcl_x.abap"), "w") as f:
|
||
f.write("CLASS zcl_x DEFINITION.\n")
|
||
res = scan_project(tmp)
|
||
self.assertEqual([(r.name, r.type) for r in res], [("ZCL_X", "class")])
|
||
|
||
def test_backward_compatible_flat_layout(self):
|
||
"""旧的扁平布局仍可扫描。"""
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
os.makedirs(os.path.join(tmp, "reports"))
|
||
with open(os.path.join(tmp, "reports", "zold.abap"), "w") as f:
|
||
f.write("REPORT zold.\n")
|
||
os.makedirs(os.path.join(tmp, "classes"))
|
||
with open(os.path.join(tmp, "classes", "zcl_old.abap"), "w") as f:
|
||
f.write("CLASS zcl_old DEFINITION.\n")
|
||
names = {r.name for r in scan_project(tmp)}
|
||
self.assertIn("ZOLD", names)
|
||
self.assertIn("ZCL_OLD", names)
|
||
|
||
def test_no_duplicate_when_both_layouts(self):
|
||
"""同一对象在两处出现只保留一次。"""
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
os.makedirs(os.path.join(tmp, "src", "TMP", "report"))
|
||
os.makedirs(os.path.join(tmp, "reports"))
|
||
for d in ("src/TMP/report", "reports"):
|
||
with open(os.path.join(tmp, *d.split("/"), "zdup.abap"), "w") as f:
|
||
f.write("REPORT zdup.\n")
|
||
res = scan_project(tmp)
|
||
self.assertEqual(len([r for r in res if r.name == "ZDUP"]), 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|
||
|