吴让宇
c06e350f6d
新增能力： - remote-enable 子命令：设置函数模块处理类型为远程启用（bgRFC/RFC 执行体必需）。 实现要点均按实机验证：PUT fmodule:processingType="rfc"（枚举值仅小写 rfc 有效， remoteEnabled/remote/RFC 等均 400）；lockHandle 走 query 参数（放 header 报 ParameterNotFound）；PUT 后 GET 复核回显；DDIC 结构参数才可远程启用（类本地类型 报"针对 RFC 不允许使用类或接口的类型"）。 - create --type functiongroup 透传 --package / --corr_nr（此前 package 被静默写成 $TMP， 导致函数组无法归目标包）。 修复： - activate 请求体补重复引用：ADT CLAS 激活端点对 objectReferences 只有 1 个引用时 返回 HTTP 200 + 空 body 且不执行激活；引用数 ≥2 才真正激活（同对象重复亦可）。 - scanner 剥离 ADT 文件后缀（xxx.clas.abap 解析错误）。 守卫测试： - tests/unit/test_bgrfc_support.py（17 例）：函数组 package 透传、processingType 助手、 remote-enable 命令（含 lockHandle 走 query/小写 rfc/错误透传/复核失配判失败）、 函数 URI 组装不得用组名冒充模块名。 - test_activate.py 增加激活请求体引用数守卫；test_repo_guards.py 增加后缀剥离守卫。 测试：tests/unit 676 + tests/test_sapcli.py 69 = 745 全绿 （PYTHONPATH=assets python -m unittest discover -s tests/unit -t tests）
948 lines
39 KiB
Python
948 lines
39 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 _make_adt_project(self, tmp: str) -> None:
|
||
"""创建 ADT/abapGit 风格项目（`<name>.<suffix>.abap`，项目仓命名规范形态）。"""
|
||
os.makedirs(os.path.join(tmp, "src", "ZINT.package", "class"))
|
||
cls_dir = os.path.join(tmp, "src", "ZINT.package", "class")
|
||
# ADT 式：zcl_demo.clas.abap
|
||
with open(os.path.join(cls_dir, "zcl_demo.clas.abap"), "w") as f:
|
||
f.write("CLASS zcl_demo DEFINITION.\n")
|
||
# 无后缀式（历史写法，仍须兼容）：zcl_plain.abap
|
||
with open(os.path.join(cls_dir, "zcl_plain.abap"), "w") as f:
|
||
f.write("CLASS zcl_plain DEFINITION.\n")
|
||
# 后缀与目录类型不匹配：不得被剥（否则对象名会被错改）
|
||
with open(os.path.join(cls_dir, "zcl_weird.prog.abap"), "w") as f:
|
||
f.write("CLASS zcl_weird DEFINITION.\n")
|
||
|
||
def test_scan_adt_style_filenames(self):
|
||
"""ADT 风格 `<name>.clas.abap` 必须解析为对象名 ZCL_DEMO（不带 .CLAS）。"""
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
self._make_adt_project(tmp)
|
||
results = scan_project(tmp)
|
||
names = {r.name for r in results}
|
||
|
||
self.assertIn("ZCL_DEMO", names)
|
||
self.assertNotIn("ZCL_DEMO.CLAS", names)
|
||
# 无后缀式兼容
|
||
self.assertIn("ZCL_PLAIN", names)
|
||
# 目录类型不匹配的后缀不剥
|
||
self.assertIn("ZCL_WEIRD.PROG", names)
|
||
|
||
def test_scan_adt_style_file_paths(self):
|
||
"""ADT 风格项目下 file 字段仍为仓内相对路径。"""
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
self._make_adt_project(tmp)
|
||
file_map = {r.name: r.file for r in scan_project(tmp)}
|
||
self.assertEqual(
|
||
file_map["ZCL_DEMO"],
|
||
"src/ZINT.package/class/zcl_demo.clas.abap",
|
||
)
|
||
|
||
def test_strip_adt_suffix_unit(self):
|
||
"""后缀剥离函数单测：只剥匹配目录类型的那一个后缀。"""
|
||
from sapcli.scanner import _strip_adt_suffix
|
||
self.assertEqual(_strip_adt_suffix("zcl_demo.clas", "class"), "zcl_demo")
|
||
self.assertEqual(_strip_adt_suffix("zcl_demo", "class"), "zcl_demo")
|
||
self.assertEqual(_strip_adt_suffix("zcl_demo.prog", "class"), "zcl_demo.prog")
|
||
self.assertEqual(_strip_adt_suffix("zrpt1.prog", "report"), "zrpt1")
|
||
self.assertEqual(_strip_adt_suffix("zdom.doma", "domain"), "zdom")
|
||
# 后缀即全部文件名时不得剥成空串
|
||
self.assertEqual(_strip_adt_suffix("clas", "class"), "clas")
|
||
|
||
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()
|
||
|