117 lines
4.4 KiB
Plaintext
117 lines
4.4 KiB
Plaintext
# -*- coding: utf-8 -*-
|
||
# 样例：复制到 ``tests/test_action_core_contract.py`` 后按实际模块改名运行。
|
||
# 默认不会被 ``python tests/run_tests.py`` 收集（扩展名为 .sample）。
|
||
"""
|
||
单业务内核、多入口契约范式（通用）。
|
||
|
||
覆盖：
|
||
- task / command adapter 调用唯一 domain service
|
||
- 兼容命令复用同一个核心函数
|
||
- CLI handler 不复制业务逻辑
|
||
- 不根据 toolbar / cron / agent 来源改变业务处理
|
||
|
||
复制后把示意函数换成真实 import；禁止按入口维护多套实现。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import unittest
|
||
from typing import Any, Callable
|
||
|
||
|
||
class DomainService:
|
||
"""唯一业务内核示意。"""
|
||
|
||
def __init__(self) -> None:
|
||
self.calls: list[dict[str, Any]] = []
|
||
|
||
def sync_records(self, *, limit: int = 100) -> dict[str, Any]:
|
||
self.calls.append({"op": "sync_records", "limit": limit})
|
||
return {"ok": True, "limit": limit, "inserted": 1}
|
||
|
||
|
||
def cmd_sync_records(domain: DomainService, *, limit: int = 100) -> dict[str, Any]:
|
||
"""task_service 编排层：可加 job_context，但不按 source 分叉业务。"""
|
||
return domain.sync_records(limit=limit)
|
||
|
||
|
||
def cmd_export_compat_refresh(domain: DomainService, *, limit: int = 100) -> dict[str, Any]:
|
||
"""兼容旧 export：刷新必须复用同一 sync 内核，禁止另写采集逻辑。"""
|
||
sync_result = domain.sync_records(limit=limit)
|
||
return {"export": "delegated-to-host-or-report", "sync": sync_result}
|
||
|
||
|
||
def cli_handle_sync(domain: DomainService, argv_limit: int) -> dict[str, Any]:
|
||
"""CLI handler：只解析参数后转发，不内嵌业务。"""
|
||
return cmd_sync_records(domain, limit=argv_limit)
|
||
|
||
|
||
def run_from_source(
|
||
domain: DomainService,
|
||
*,
|
||
source: str,
|
||
limit: int = 10,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
宿主可能传入 source=toolbar|cron|agent|skill-detail。
|
||
正确实现：忽略 source，业务结果一致。
|
||
"""
|
||
_ = source # 不得用于分支业务语义
|
||
return cmd_sync_records(domain, limit=limit)
|
||
|
||
|
||
class TestActionCoreContractSample(unittest.TestCase):
|
||
def test_task_adapter_calls_single_domain_service(self) -> None:
|
||
domain = DomainService()
|
||
out = cmd_sync_records(domain, limit=5)
|
||
self.assertTrue(out["ok"])
|
||
self.assertEqual(domain.calls, [{"op": "sync_records", "limit": 5}])
|
||
|
||
def test_compat_command_reuses_same_core(self) -> None:
|
||
domain = DomainService()
|
||
cmd_export_compat_refresh(domain, limit=3)
|
||
self.assertEqual(len(domain.calls), 1)
|
||
self.assertEqual(domain.calls[0]["op"], "sync_records")
|
||
|
||
def test_cli_handler_does_not_duplicate_business_logic(self) -> None:
|
||
domain = DomainService()
|
||
# CLI 与 task 编排应落到同一函数对象（示意）
|
||
self.assertIs(cli_handle_sync.__wrapped__ if hasattr(cli_handle_sync, "__wrapped__") else None, None)
|
||
out = cli_handle_sync(domain, 7)
|
||
self.assertEqual(out["limit"], 7)
|
||
self.assertEqual(domain.calls[-1], {"op": "sync_records", "limit": 7})
|
||
|
||
# 强化：CLI 路径与 cmd 路径结果一致
|
||
domain2 = DomainService()
|
||
a = cli_handle_sync(domain2, 2)
|
||
domain3 = DomainService()
|
||
b = cmd_sync_records(domain3, limit=2)
|
||
self.assertEqual(a, b)
|
||
|
||
def test_business_result_independent_of_entry_source(self) -> None:
|
||
domain = DomainService()
|
||
results = [
|
||
run_from_source(domain, source=src, limit=4)
|
||
for src in ("toolbar", "cron", "agent", "skill-detail")
|
||
]
|
||
self.assertEqual(results[0], results[1])
|
||
self.assertEqual(results[1], results[2])
|
||
self.assertEqual(results[2], results[3])
|
||
self.assertEqual(len(domain.calls), 4)
|
||
self.assertTrue(all(c["op"] == "sync_records" and c["limit"] == 4 for c in domain.calls))
|
||
|
||
def test_entrypoint_wiring_points_to_same_callable(self) -> None:
|
||
"""示意：manifest 多 placements 应对应同一 Python 入口函数。"""
|
||
domain = DomainService()
|
||
handlers: dict[str, Callable[..., dict[str, Any]]] = {
|
||
"toolbar": lambda: cmd_sync_records(domain, limit=1),
|
||
"cron": lambda: cmd_sync_records(domain, limit=1),
|
||
"agent": lambda: cmd_sync_records(domain, limit=1),
|
||
}
|
||
outs = [handlers[k]() for k in handlers]
|
||
self.assertEqual(outs[0], outs[1])
|
||
self.assertEqual(outs[1], outs[2])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|