diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 0f58be451..639d267c4 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -125,22 +125,18 @@ jobs: fail-fast: false matrix: include: - - shard: '1/8' + - shard: '1/6' index: '1' - - shard: '2/8' + - shard: '2/6' index: '2' - - shard: '3/8' + - shard: '3/6' index: '3' - - shard: '4/8' + - shard: '4/6' index: '4' - - shard: '5/8' + - shard: '5/6' index: '5' - - shard: '6/8' + - shard: '6/6' index: '6' - - shard: '7/8' - index: '7' - - shard: '8/8' - index: '8' steps: - name: Checkout code @@ -160,7 +156,7 @@ jobs: - name: Generate coverage data timeout-minutes: 10 - run: uv run --locked --no-sync python -m coverage run --parallel-mode tests/run.py --shard "${{ matrix.shard }}" + run: uv run --locked --no-sync python -m coverage run --parallel-mode tests/run.py --exclude-architecture-gate --shard "${{ matrix.shard }}" - name: Upload coverage data uses: actions/upload-artifact@v7 diff --git a/docs/testing.md b/docs/testing.md index 8a9cf6eb8..bc0d9a9a1 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -15,13 +15,16 @@ uv run --locked --no-sync python tests/run.py --serial # 串行全 uv run --locked --no-sync python tests/run.py --shard 1/4 # 只跑指定分片，供 CI 复用 ``` -`tests/run.py` 的 runner 参数只有 `--serial` 和 `--shard N/TOTAL`；其余参数保持原顺序 +`tests/run.py` 的 runner 参数只有 `--serial`、`--shard N/TOTAL` 和 `--exclude-architecture-gate`；其余参数保持原顺序 透传给 pytest，例如 `python tests/run.py -q --maxfail=1`。分片使用受版本控制的 `tests/fixtures/durations.json` 慢文件耗时估算：按耗时降序，逐个分给预计总耗时最短的分片； 耗时相同时按路径、分片编号决定归属，片内仍按路径排序。新增或未记录文件使用 1 秒权重， 不依赖本机缓存，确保同一代码版本在本地与 CI 的文件集合和顺序一致。 耗时数据来自成功 Ubuntu/Python 3.14 Coverage 日志的逐文件进度时间戳（含续行）， 只记录至少 3 秒的文件并向上取整；后续发现慢文件偏移时可依据成功 CI 日志更新。 +`--exclude-architecture-gate` 跳过 `ARCHITECTURE_GATE_TESTS` 列出的架构测试文件，这些文件由 CI 的 +`architecture` job 在同一进程内运行并共享全量扫描结果（`tests/architecture_cache.py`），覆盖率分片不再重复执行； +本地默认全量仍包含它们。 这些估算仅用于调度，不构成性能或覆盖率基线。 - 不再使用 `python -m unittest discover`：它不导入 `tests` 包、收不到纯函数用例，且绕过 `conftest.py` 的隔离。 @@ -183,9 +186,9 @@ def test_recognize_prefers_explicit_identity(sample_meta, monkeypatch): - **合并检查复用**：单测/架构与 Pylint 工作流各自保留 PR、push 和手动入口。PR 完整通过全部门禁后，末尾 `CI proof ()` job 记录实际验证的模拟合并提交。合并 push 只在同一工作流的最新 PR 运行完整成功、证明成功、代码树完全相同、模拟合并父提交分别等于 push 前的目标分支和 PR head 时跳过重复门禁。直接 push、强制 push、基线变化、旧工作流缺少证明、失败/未完成运行或 API 异常都执行全量；不依赖提交消息，也不把 PR 的 head SHA 当作测试的合并 SHA。标准 merge/squash/rebase 仅在上述证据一致时复用。GitHub 构建与发布工作流保持独立。 - **去重脚本验证**：`node --test .github/scripts/reuse.test.mjs` 离线覆盖成功复用及保守回退，两个检查工作流均在判定前执行。复用 job 仅持有 contents/actions/pull-requests 读取权限；没有修改分支保护设置。 -- **门禁**：`.github/workflows/test.yml` 在指向 `v3` 的 `pull_request` / `push` 及手动触发时，从 `uv.lock` 同步环境。独立 `architecture` job 先运行宿主依赖、运行契约和基线 CLI 快速门禁；全量测试通过 `coverage run --parallel-mode tests/run.py --shard N/8` 分到 8 个 job，一次执行同时验证单测并采集覆盖率。每个分片都有独立进程和临时 `CONFIG_DIR`，不共用 SQLite 或进程级状态，由单一报告 job 合并后检查 Application 与 Domain 的固定 80% 基线。 +- **门禁**：`.github/workflows/test.yml` 在指向 `v3` 的 `pull_request` / `push` 及手动触发时，从 `uv.lock` 同步环境。独立 `architecture` job 先运行宿主依赖、运行契约和基线 CLI 快速门禁；其余测试通过 `coverage run --parallel-mode tests/run.py --exclude-architecture-gate --shard N/6` 分到 6 个 job，一次执行同时验证单测并采集覆盖率；门禁已运行的架构测试不在分片中重复执行。每个分片都有独立进程和临时 `CONFIG_DIR`，不共用 SQLite 或进程级状态，由单一报告 job 合并后检查 Application 与 Domain 的固定 80% 基线。 - **跨仓观察**：`.github/workflows/architecture-observe.yml` 每周或手工检出官方插件仓最新 `main`，使用 `--check-plugins` 比较公开导入、Hook 和动态 API 契约。它只上传 `official-plugin-architecture-report.json`，不会自动刷新 fixture；语义变化必须人工审查后显式执行 `--write-plugins`。 - **静态检查**：`.github/workflows/pylint.yml` 对指向 `v3` 的 PR、推送和手工触发运行 Pylint。PR/推送改动到的 Python 文件是硬门禁；`app/` 全量扫描保留为建议性 JSON 构建工件，存量告警不会掩盖或阻塞本次增量治理。 - **PR 本地验证**：按上文「验证范围与维护者安排」选择 contributor 默认检查或已记录的维护者安排，统一处理证据复用与失败归属；需要断点、输出顺序或测试污染诊断时使用 `--serial`。确认受影响路径与零真实出站，准确标注验证范围；本地执行安排不改变 CI 的全量验证与复用合同。 -- **覆盖率门禁**：`Unit Tests with Coverage` jobs 会在 `v3` 的 PR、push 和手工触发中通过 `tests/run.py --shard N/8` 并行采集覆盖率数据，`Coverage Report` 再合并全部分片并只读检查 Application 与 Domain 是否达到 Ubuntu/Python 3.14 canonical 的固定 80% 行覆盖率基线，同时上传 JSON / XML 工件。低于 80% 会阻塞；达到或超过 80% 不要求同步运行时语句计数。macOS 本地报告只用于诊断，不直接作为可提交基线。 +- **覆盖率门禁**：`Unit Tests with Coverage` jobs 会在 `v3` 的 PR、push 和手工触发中通过 `tests/run.py --exclude-architecture-gate --shard N/6` 并行采集覆盖率数据，`Coverage Report` 再合并全部分片并只读检查 Application 与 Domain 是否达到 Ubuntu/Python 3.14 canonical 的固定 80% 行覆盖率基线，同时上传 JSON / XML 工件。低于 80% 会阻塞；达到或超过 80% 不要求同步运行时语句计数。macOS 本地报告只用于诊断，不直接作为可提交基线。 - 复现 CI 使用 `uv sync --locked`；主程序运行依赖位于 `[project].dependencies`，pytest 与覆盖率工具位于默认 `dev` 依赖组。 diff --git a/tests/architecture_cache.py b/tests/architecture_cache.py new file mode 100644 index 000000000..6bf35d531 --- /dev/null +++ b/tests/architecture_cache.py @@ -0,0 +1,47 @@ +"""架构测试共享的全量扫描结果：同一 pytest 进程内每类扫描只执行一次。 + +门禁把多个架构测试文件放在同一进程运行，这些文件会对同一份源码重复做相同的全量收集。 +缓存只存放原始结果，每次返回深拷贝，测试修改返回值不会影响其他用例。 +""" + +from __future__ import annotations + +import ast +import copy +import functools +from pathlib import Path +from typing import Any + +from scripts.architecture.baseline import collect_current_event_facts, collect_dependency_baseline + + +@functools.cache +def _dependency_baseline() -> dict[str, Any]: + return collect_dependency_baseline() + + +@functools.cache +def _current_event_facts() -> dict[str, list[dict[str, Any]]]: + return collect_current_event_facts() + + +def dependency_baseline() -> dict[str, Any]: + """返回当前宿主依赖基线的独立副本。""" + return copy.deepcopy(_dependency_baseline()) + + +def current_event_facts() -> dict[str, list[dict[str, Any]]]: + """返回当前宿主事件事实的独立副本。""" + return copy.deepcopy(_current_event_facts()) + + +@functools.cache +def parse_module(path: Path) -> ast.Module: + """解析并缓存源码 AST；调用方只读取，不得修改返回的节点。""" + return ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + + +@functools.cache +def walk_module(tree: ast.Module) -> tuple[ast.AST, ...]: + """按 ``ast.walk`` 顺序返回模块全部节点，供多个测试重复遍历同一只读 AST。""" + return tuple(ast.walk(tree)) diff --git a/tests/fixtures/durations.json b/tests/fixtures/durations.json index 71dfafd01..49e1be8e1 100644 --- a/tests/fixtures/durations.json +++ b/tests/fixtures/durations.json @@ -1,37 +1,39 @@ { - "source": "https://github.com/jxxghp/MoviePilot/actions/runs/34326739404", - "description": "Ubuntu/Python 3.14 coverage 日志中逐文件进度时间戳的耗时估算，含续行；仅记录至少 3 秒的文件并向上取整，其余文件使用 1 秒权重。用于分片均衡，不是性能或覆盖率基线。", + "source": "https://github.com/jxxghp/MoviePilot/actions/runs/36409216765", + "description": "Ubuntu/Python 3.14 coverage 日志中逐文件进度时间戳的耗时估算，含续行；仅记录至少 3 秒的文件并向上取整，其余文件使用 1 秒权重。不含架构门禁 job 运行的文件；用于分片均衡，不是性能或覆盖率基线。", "files": { - "test_agent_api_lazy_imports.py": 15, - "test_agent_api_surface_audit.py": 5, - "test_agent_lazy_runtime_boundary.py": 7, - "test_agent_tool_result_policy.py": 4, - "test_api_response.py": 29, - "test_architecture_adapter_imports.py": 67, - "test_architecture_baseline_cli.py": 7, - "test_architecture_contract_baseline.py": 6, - "test_architecture_dependencies.py": 95, - "test_architecture_egress.py": 67, - "test_architecture_event_facts.py": 17, - "test_architecture_event_policy.py": 16, - "test_bangumi_media_type.py": 4, - "test_chain_base_boundary.py": 9, - "test_data_cleanup_chain.py": 7, - "test_duplicate_code.py": 6, - "test_execute_command_tool.py": 4, - "test_host_runtime_context.py": 4, + "test_module_manager_capability_adapter.py": 38, + "test_agent_evaluation_runtime.py": 37, + "test_api_response.py": 22, + "test_router_aggregation.py": 20, + "test_agent_api_lazy_imports.py": 16, + "test_chain_base_boundary.py": 14, + "test_agent_lazy_runtime_boundary.py": 8, + "test_agent_evaluation_server.py": 7, + "test_service_locator_gate.py": 7, + "test_subscription_governance_scale.py": 7, + "test_transfer_history_architecture.py": 7, + "test_subscription_search_queue.py": 6, + "test_subscription_search_governance.py": 6, + "test_download_architecture.py": 6, + "test_duplicate_code.py": 5, + "test_optional_sdk_lazy_import.py": 5, + "test_workflow_mutation_command.py": 5, + "test_chain_layering.py": 5, + "test_agent_subagent_permissions.py": 5, "test_legacy_import_compat.py": 5, - "test_llm_facade_governance.py": 4, - "test_main_direct_execution.py": 4, - "test_media_auxiliary.py": 4, - "test_module_manager_capability_adapter.py": 36, - "test_monitor_mount_isolation.py": 5, - "test_music_matching.py": 6, - "test_router_aggregation.py": 10, - "test_service_locator_gate.py": 5, - "test_subscription_governance_scale.py": 6, - "test_subscription_search_governance.py": 4, + "test_monitor_mount_isolation.py": 4, + "test_data_locator_architecture.py": 4, + "test_workflow_invoke_plugin.py": 4, + "test_agent_api_surface_audit.py": 4, + "test_scheduler_architecture.py": 4, + "test_runtime_execution.py": 4, + "test_zspace_sync_items.py": 4, + "test_docker_bootstrap.py": 4, "test_task_ownership_gate.py": 4, - "test_transfer_history_architecture.py": 6 + "test_host_runtime_context.py": 4, + "test_workflow_runtime_facade.py": 4, + "test_agent_tool_result_policy.py": 4, + "test_main_direct_execution.py": 4 } } diff --git a/tests/run.py b/tests/run.py index 7940cdf00..844018e08 100644 --- a/tests/run.py +++ b/tests/run.py @@ -15,11 +15,22 @@ RUNNER_PATH = Path(__file__).resolve() DEFAULT_SHARD_COUNT = 4 DURATIONS_PATH = TESTS_DIR / "fixtures" / "durations.json" DEFAULT_TEST_DURATION = 1.0 +# CI 架构门禁 job 独立运行的测试文件；分片使用 --exclude-architecture-gate 时不再重复执行 +ARCHITECTURE_GATE_TESTS = ( + "test_architecture_dependencies.py", + "test_architecture_adapter_imports.py", + "test_architecture_egress.py", + "test_architecture_event_facts.py", + "test_architecture_event_policy.py", + "test_architecture_contract_baseline.py", + "test_architecture_baseline_cli.py", +) -def collect_test_files() -> list[Path]: - """按稳定路径顺序返回根测试目录中的全部测试文件。""" - return sorted(TESTS_DIR.glob("test_*.py")) +def collect_test_files(exclude_architecture_gate: bool = False) -> list[Path]: + """按稳定路径顺序返回根测试目录中的测试文件，可排除已由架构门禁运行的文件。""" + excluded = set(ARCHITECTURE_GATE_TESTS) if exclude_architecture_gate else set() + return sorted(path for path in TESTS_DIR.glob("test_*.py") if path.name not in excluded) def load_test_durations() -> dict[str, float]: @@ -77,9 +88,16 @@ def parse_args(argv: Sequence[str]) -> tuple[argparse.Namespace, list[str]]: metavar="N/TOTAL", help="只运行指定文件分片；CI 使用同一参数启动独立 job。", ) + parser.add_argument( + "--exclude-architecture-gate", + action="store_true", + help="跳过 CI 架构门禁已运行的测试文件，供覆盖率分片使用。", + ) args, pytest_args = parser.parse_known_args(argv) if args.serial and args.shard is not None: parser.error("--serial 不能与 --shard 同时使用") + if args.serial and args.exclude_architecture_gate: + parser.error("--serial 不能与 --exclude-architecture-gate 同时使用") return args, pytest_args @@ -89,20 +107,23 @@ def run_pytest(paths: Sequence[Path], pytest_args: Sequence[str]) -> int: def _worker_command( - shard_index: int, shard_count: int, pytest_args: Sequence[str] + shard_index: int, shard_count: int, pytest_args: Sequence[str], + exclude_architecture_gate: bool = False, ) -> list[str]: - """构造与 CI 完全相同的单分片 worker 命令。""" + """构造与 CI 完全相同的单分片 worker 命令；排除选项必须一并传递，否则分片划分不一致。""" return [ sys.executable, str(RUNNER_PATH), "--shard", f"{shard_index}/{shard_count}", + *(["--exclude-architecture-gate"] if exclude_architecture_gate else []), *pytest_args, ] def run_parallel_shards( - shards: Sequence[Sequence[Path]], pytest_args: Sequence[str] + shards: Sequence[Sequence[Path]], pytest_args: Sequence[str], + exclude_architecture_gate: bool = False, ) -> int: """启动独立 pytest 进程并等待全部文件分片结束。""" shard_count = len(shards) @@ -116,7 +137,9 @@ def run_parallel_shards( ) processes.append(( shard_index, - subprocess.Popen(_worker_command(shard_index, shard_count, pytest_args)), + subprocess.Popen(_worker_command( + shard_index, shard_count, pytest_args, exclude_architecture_gate, + )), )) exit_code = 0 @@ -146,7 +169,7 @@ def main(argv: Sequence[str] | None = None) -> int: if args.serial: return run_pytest([TESTS_DIR], pytest_args) - test_files = collect_test_files() + test_files = collect_test_files(args.exclude_architecture_gate) if not test_files: print(f"未在 {TESTS_DIR} 找到 test_*.py", file=sys.stderr) return 2 @@ -164,7 +187,7 @@ def main(argv: Sequence[str] | None = None) -> int: return run_pytest(selected, pytest_args) shards = split_test_files(test_files, DEFAULT_SHARD_COUNT) - return run_parallel_shards(shards, pytest_args) + return run_parallel_shards(shards, pytest_args, args.exclude_architecture_gate) if __name__ == "__main__": diff --git a/tests/test_architecture_adapter_imports.py b/tests/test_architecture_adapter_imports.py index 245cfea0c..6852988e6 100644 --- a/tests/test_architecture_adapter_imports.py +++ b/tests/test_architecture_adapter_imports.py @@ -5,9 +5,9 @@ from collections import Counter from pathlib import Path from scripts.architecture.baseline import ( - collect_dependency_baseline, collect_direct_adapter_imports, ) +from tests.architecture_cache import dependency_baseline PROJECT_ROOT = Path(__file__).parents[1] DEPENDENCY_POLICY_PATH = ( @@ -208,7 +208,7 @@ def test_direct_adapter_collector_handles_relative_imports_and_scope( def test_current_direct_adapter_imports_are_stable_generated_facts() -> None: """生成事实必须有稳定排序、自洽统计，且不保存符号或行号。""" - contract = collect_dependency_baseline()["direct_adapter_imports"] + contract = dependency_baseline()["direct_adapter_imports"] edges = contract["edges"] sources = sorted({edge["source"] for edge in edges}) targets = sorted({edge["target"] for edge in edges}) @@ -237,7 +237,7 @@ def test_current_direct_adapter_imports_are_stable_generated_facts() -> None: def test_current_direct_adapter_imports_match_temporary_debt_policy() -> None: """现存直连必须逐条绑定冻结 owner，并允许债务集合只减不增。""" policy = json.loads(DEPENDENCY_POLICY_PATH.read_text(encoding="utf-8")) - contract = collect_dependency_baseline()["direct_adapter_imports"] + contract = dependency_baseline()["direct_adapter_imports"] adapter_policy = policy["direct_adapter_imports"] entries = adapter_policy["entries"] diff --git a/tests/test_architecture_ci.py b/tests/test_architecture_ci.py index a61b26f6c..b1f499539 100644 --- a/tests/test_architecture_ci.py +++ b/tests/test_architecture_ci.py @@ -5,6 +5,8 @@ from pathlib import Path from ruamel.yaml import YAML +from tests.run import ARCHITECTURE_GATE_TESTS + PROJECT_ROOT = Path(__file__).parents[1] WORKFLOW_ROOT = PROJECT_ROOT / ".github" / "workflows" @@ -64,6 +66,13 @@ def test_unit_test_workflow_has_independent_host_architecture_gate(): assert "scripts/architecture/mypy_ratchet.py --write" not in commands assert "scripts/architecture/service_locator.py" in commands assert "scripts/startup/performance.py --check --repeat 3" in commands + # 覆盖率分片跳过的文件必须恰好由门禁运行，否则会有测试在 CI 中无人执行 + gate_test_files = { + token.removeprefix("tests/") + for token in commands.split() + if token.startswith("tests/test_") and token.endswith(".py") + } + assert gate_test_files == set(ARCHITECTURE_GATE_TESTS) def test_official_plugin_observation_is_scheduled_and_never_writes_fixture(): @@ -98,17 +107,8 @@ def test_coverage_jobs_parallelize_data_and_keep_one_global_ratchet() -> None: assert shard_job["name"] == "Unit Tests with Coverage (${{ matrix.shard }})" assert shard_job["timeout-minutes"] == 15 shard_matrix = shard_job["strategy"]["matrix"]["include"] - assert [item["shard"] for item in shard_matrix] == [ - "1/8", - "2/8", - "3/8", - "4/8", - "5/8", - "6/8", - "7/8", - "8/8", - ] - assert [item["index"] for item in shard_matrix] == [str(index) for index in range(1, 9)] + assert [item["shard"] for item in shard_matrix] == [f"{index}/6" for index in range(1, 7)] + assert [item["index"] for item in shard_matrix] == [str(index) for index in range(1, 7)] assert report_job["needs"] == "coverage-shard" assert report_job["runs-on"] == "ubuntu-latest" assert report_job["timeout-minutes"] == 10 @@ -121,7 +121,10 @@ def test_coverage_jobs_parallelize_data_and_keep_one_global_ratchet() -> None: step for step in shard_steps if step.get("name") == "Generate coverage data" ) assert generate_step["timeout-minutes"] == 10 - assert "coverage run --parallel-mode tests/run.py --shard" in generate_step["run"] + assert ( + "coverage run --parallel-mode tests/run.py --exclude-architecture-gate --shard" + in generate_step["run"] + ) upload_data_step = next( step for step in shard_steps if step.get("name") == "Upload coverage data" ) diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index 0c27148e5..a9a263083 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -4,9 +4,6 @@ from functools import lru_cache from pathlib import Path from app.runtime.compat.manifest import MODULE_ALIASES, SYMBOL_ALIASES -from scripts.architecture.baseline import ( - collect_current_event_facts as _collect_current_event_facts, -) from scripts.architecture.baseline import ( discover_modules as _discover_modules, ) @@ -16,6 +13,11 @@ from scripts.architecture.baseline import ( from scripts.architecture.baseline import ( strongly_connected_components as _strongly_connected_components, ) +from tests.architecture_cache import ( + current_event_facts as _collect_current_event_facts, +) +from tests.architecture_cache import parse_module as _parse_source +from tests.architecture_cache import walk_module as _walk_module PROJECT_ROOT = Path(__file__).parents[1] APP_ROOT = PROJECT_ROOT / "app" @@ -212,9 +214,9 @@ FORBIDDEN_IMPORT_PREFIXES = { def _legacy_imports(path: Path) -> set[str]: """提取源码中的静态和常量动态旧路径导入。""" - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) imports: set[str] = set() - for node in ast.walk(tree): + for node in _walk_module(tree): candidates: list[str] = [] if isinstance(node, ast.Import): candidates.extend(alias.name for alias in node.names) @@ -251,7 +253,7 @@ def _compat_symbol_references(tree: ast.AST) -> set[tuple[int, str]]: module_bindings: dict[str, str] = {} references: set[tuple[int, str]] = set() - for node in ast.walk(tree): + for node in _walk_module(tree): if isinstance(node, ast.Import): for alias in node.names: binding = alias.asname or alias.name.split(".", maxsplit=1)[0] @@ -264,7 +266,7 @@ def _compat_symbol_references(tree: ast.AST) -> set[tuple[int, str]]: binding = alias.asname or alias.name module_bindings[binding] = f"{node.module}.{alias.name}" - for node in ast.walk(tree): + for node in _walk_module(tree): if not isinstance(node, ast.Attribute): continue parts = _attribute_parts(node) @@ -282,7 +284,7 @@ def _compat_symbol_references(tree: ast.AST) -> set[tuple[int, str]]: def _class_annotations(path: Path, class_name: str) -> dict[str, str]: """返回指定类的源码级字段注解。""" - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) class_node = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == class_name) return { node.target.id: ast.unparse(node.annotation) @@ -373,8 +375,8 @@ def test_runtime_dependencies_use_same_named_single_word_package() -> None: relative = path.relative_to(APP_ROOT) if relative.parts[0] == "plugins": continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if isinstance(node, ast.ImportFrom) and node.module == "app.runtime.dependencies": violations.append(f"{relative}:{node.lineno}:from-package-root") elif isinstance(node, ast.Import): @@ -416,8 +418,8 @@ def test_domain_classification_is_a_pure_direct_import_package() -> None: ) violations: list[str] = [] for path in package.glob("*.py"): - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): imported_modules: list[str] = [] if isinstance(node, ast.ImportFrom) and node.module: imported_modules.append(node.module) @@ -434,8 +436,8 @@ def test_domain_classification_is_a_pure_direct_import_package() -> None: for path in APP_ROOT.rglob("*.py"): if path.is_relative_to(APP_ROOT / "plugins"): continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if isinstance(node, ast.ImportFrom) and node.module == "app.domain.classification": root_imports.append(f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}") elif isinstance(node, ast.Import): @@ -492,8 +494,8 @@ def test_application_torrent_uses_same_named_single_word_package() -> None: for path in root.rglob("*.py"): if path.is_relative_to(APP_ROOT / "plugins"): continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if isinstance(node, ast.ImportFrom) and node.module == "app.application.torrent": root_imports.append(f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}") assert root_imports == [] @@ -510,15 +512,15 @@ def test_host_module_package_roots_only_export_capability_entrypoints() -> None: for path in (APP_ROOT / "modules").rglob("*.py"): if path.is_relative_to(APP_ROOT / "plugins"): continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if isinstance(node, ast.ImportFrom) and node.module == "app.modules._base": root_imports.append(f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}") assert root_imports == [] for package_name, expected_exports in HOST_MODULE_PACKAGE_EXPORTS.items(): path = APP_ROOT / "modules" / package_name / "__init__.py" - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + tree = _parse_source(path) assignments = { target.id: ast.literal_eval(node.value) for node in tree.body @@ -588,8 +590,8 @@ def test_domain_media_projection_uses_single_word_owner_package() -> None: relative = path.relative_to(APP_ROOT) if relative.parts[0] == "plugins" or path == context_path: continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if not isinstance(node, ast.Attribute) or not isinstance(node.value, ast.Name): continue if node.value.id == "MediaInfo" and node.attr in { @@ -605,7 +607,7 @@ def test_domain_media_projection_uses_single_word_owner_package() -> None: def test_workflow_query_contract_returns_only_typed_snapshots(): """工作流正式查询端口不得退化为 Any 或 ORM 返回值。""" path = APP_ROOT / "application" / "workflow.py" - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + tree = _parse_source(path) query_classes = { node.name: node for node in tree.body @@ -1085,10 +1087,10 @@ def test_runtime_composition_is_the_sole_host_and_domain_runtime_owner(): for path in APP_ROOT.rglob("*.py"): if path.is_relative_to(APP_ROOT / "plugins"): continue - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + tree = _parse_source(path) if any( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "HostRuntime" - for node in ast.walk(tree) + for node in _walk_module(tree) ): constructors.append(path.relative_to(APP_ROOT).as_posix()) assert constructors == ["startup/composition/runtime.py"] @@ -1218,8 +1220,8 @@ def test_plugin_route_refresh_is_imported_from_its_application_owner(): relative = path.relative_to(APP_ROOT) if relative.parts[0] == "plugins": continue - tree = ast.parse(path.read_text(encoding="utf-8")) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if not isinstance(node, ast.ImportFrom) or node.level: continue if node.module == owner: @@ -1299,7 +1301,7 @@ def test_host_code_does_not_use_compat_symbol_aliases() -> None: or relative.parts[:2] == ("sdk", "_legacy") ): continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) violations.extend( f"{relative.as_posix()}:{line}:{symbol}" for line, symbol in sorted(_compat_symbol_references(tree)) ) @@ -1321,8 +1323,8 @@ def test_host_code_uses_explicit_runtime_facade_getters(): "compat", ): continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if isinstance(node, ast.Assign) and isinstance(node.value, ast.Name): target_names = {target.id for target in node.targets if isinstance(target, ast.Name)} if "SystemConfigOper" in target_names and node.value.id == "get_configured_system_config": @@ -1371,8 +1373,8 @@ def test_user_chain_and_agent_ports_are_typed_and_orm_free(): ] violations: list[str] = [] for path in production_paths: - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if isinstance(node, ast.ClassDef) and node.name == "UserOper": violations.append(f"{path.relative_to(PROJECT_ROOT)}:{node.lineno}:class") if isinstance(node, ast.ImportFrom) and node.module == "app.db.oper.user": @@ -1411,8 +1413,8 @@ def test_transfer_pending_oper_import_is_confined_to_database_boundary(): relative = path.relative_to(PROJECT_ROOT).as_posix() if relative.startswith("app/plugins/") or relative in allowed_paths: continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if isinstance(node, ast.Import): if any(alias.name == "app.db.oper.transferpending" for alias in node.names): violations.append(f"{relative}:{node.lineno}") @@ -1431,16 +1433,16 @@ def test_transfer_pending_oper_import_is_confined_to_database_boundary(): def test_startup_injects_transactional_transfer_admission_repository(): """启动组合根必须向 Chain 注入事务型整理准入仓储。""" path = APP_ROOT / "startup" / "composition" / "chain.py" - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) imports_repository = any( isinstance(node, ast.ImportFrom) and node.module == "app.db.adapters.transfer.admission" and any(alias.name == "TransactionalTransferAdmissionRepository" for alias in node.names) - for node in ast.walk(tree) + for node in _walk_module(tree) ) chain_context_calls = [ node - for node in ast.walk(tree) + for node in _walk_module(tree) if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "ChainRuntimeContext" ] @@ -1480,10 +1482,10 @@ def test_scheduler_does_not_depend_on_database_implementation(): def test_monitor_dispatcher_uses_explicit_history_port_getter(): """监控分发器不得把兼容 TransferHistoryPort 伪装成数据库 Oper。""" path = APP_ROOT / "monitor" / "dispatcher.py" - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) violations = [ f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}" - for node in ast.walk(tree) + for node in _walk_module(tree) if isinstance(node, ast.ImportFrom) and node.module == "app.application.history" and any(alias.name == "TransferHistoryPort" for alias in node.names) @@ -1505,8 +1507,8 @@ def test_canonical_service_config_consumers_use_application_directory(): ] violations: list[str] = [] for path in paths: - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if isinstance(node, ast.ImportFrom) and node.module == "app.runtime.extensions.service": violations.append(f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}") if ( @@ -1526,7 +1528,7 @@ def test_plugin_components_do_not_reexport_legacy_abi_names(): manager_path = APP_ROOT / "runtime" / "extensions" / "plugin" / "manager.py" for root in PLUGIN_COMPONENT_ROOTS: for path in (PROJECT_ROOT / root).rglob("*.py"): - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) for node in tree.body: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): if node.name == "__getattr__" or node.name in PLUGIN_LEGACY_ABI_NAMES: @@ -1562,8 +1564,8 @@ def test_host_code_uses_precise_schema_modules(): relative = path.relative_to(APP_ROOT) if relative.parts[0] in {"plugins", "schemas"}: continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if isinstance(node, ast.ImportFrom) and node.module == "app": if any(alias.name == "schemas" for alias in node.names): violations.append(str(relative)) @@ -1580,9 +1582,9 @@ def test_database_internals_do_not_import_db_facades(): for path in (APP_ROOT / "db").rglob("*.py"): if path.name == "__init__.py": continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) if any( - isinstance(node, ast.ImportFrom) and node.module in {"app.db", "app.db.models"} for node in ast.walk(tree) + isinstance(node, ast.ImportFrom) and node.module in {"app.db", "app.db.models"} for node in _walk_module(tree) ): violations.append(str(path.relative_to(PROJECT_ROOT))) assert violations == [] @@ -1593,8 +1595,8 @@ def test_database_opers_use_dboper_transaction_dispatchers(): runner_names = {"run_sync_transaction", "run_async_transaction"} violations: list[str] = [] for path in (APP_ROOT / "db" / "oper").rglob("*.py"): - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if not isinstance(node, ast.ImportFrom) or node.module != "app.db.uow": continue imported = {alias.name for alias in node.names} & runner_names @@ -1621,9 +1623,9 @@ def test_models_and_base_require_explicit_database_sessions(): paths = [APP_ROOT / "db" / "base.py"] paths.extend((APP_ROOT / "db" / "models").rglob("*.py")) for path in paths: - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) relative = str(path.relative_to(PROJECT_ROOT)) - nodes = list(ast.walk(tree)) + nodes = list(_walk_module(tree)) for node in nodes: if isinstance(node, ast.ImportFrom) and node.module == "app.db.decorators": violations.append(f"{relative}:{node.lineno}:decorator-import") @@ -1662,8 +1664,8 @@ def test_plugin_sdk_does_not_import_or_export_host_models(): violations: list[str] = [] for path in (APP_ROOT / "sdk").rglob("*.py"): relative = path.relative_to(PROJECT_ROOT).as_posix() - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if ( isinstance(node, ast.ImportFrom) and node.module @@ -1753,9 +1755,9 @@ def test_application_does_not_import_transport_frameworks(): """应用层不得依赖 FastAPI、Starlette 或宿主 HTTP 适配器。""" violations: dict[str, set[str]] = {} for path in (APP_ROOT / "application").rglob("*.py"): - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) forbidden: set[str] = set() - for node in ast.walk(tree): + for node in _walk_module(tree): candidates: list[str] = [] if isinstance(node, ast.Import): candidates.extend(alias.name for alias in node.names) @@ -1790,8 +1792,8 @@ def test_host_code_does_not_use_string_utils_facade(): relative = path.relative_to(APP_ROOT) if relative.parts[0] in {"plugins", "sdk"}: continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - if any(isinstance(node, ast.Name) and node.id == "StringUtils" for node in ast.walk(tree)): + tree = _parse_source(path) + if any(isinstance(node, ast.Name) and node.id == "StringUtils" for node in _walk_module(tree)): violations.append(str(relative)) assert violations == [] @@ -1804,8 +1806,8 @@ def test_host_code_does_not_import_chain_plugin_abi_roots() -> None: relative = path.relative_to(APP_ROOT) if relative.parts[0] == "plugins": continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if isinstance(node, ast.ImportFrom) and node.module in compatibility_roots: violations.append(f"{relative}:{node.lineno}:{node.module}") elif isinstance(node, ast.Import): @@ -1833,8 +1835,8 @@ def test_foundation_does_not_emit_runtime_logs(): """基础机制不打印或初始化日志系统，运行期诊断由上层调用方负责。""" violations: list[str] = [] for path in (APP_ROOT / "foundation").rglob("*.py"): - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if isinstance(node, ast.Import) and any(alias.name == "logging" for alias in node.names): violations.append(str(path.relative_to(PROJECT_ROOT))) break @@ -1880,12 +1882,12 @@ def test_passkey_application_does_not_select_cache_backend(): def test_startup_explicitly_configures_passkey_challenge_cache(): """PassKey challenge 缓存必须由启动组合根显式装配。""" path = APP_ROOT / "startup" / "composition" / "security.py" - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) configured = any( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "configure_passkey_challenge_cache" - for node in ast.walk(tree) + for node in _walk_module(tree) ) assert configured is True @@ -1947,8 +1949,8 @@ def test_chain_does_not_import_downloader_sdks(): forbidden_sdks = {"qbittorrentapi", "transmission_rpc"} violations: list[str] = [] for path in (APP_ROOT / "chain").rglob("*.py"): - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): names: list[str] = [] if isinstance(node, ast.Import): names.extend(alias.name for alias in node.names) @@ -2110,10 +2112,10 @@ def test_host_consumers_get_agent_manager_through_application_facade(): for module_name, path in _discover_modules().items(): if module_name.startswith(("app.agent", "app.startup")): continue - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + tree = _parse_source(path) imported = { alias.name - for node in ast.walk(tree) + for node in _walk_module(tree) if isinstance(node, ast.ImportFrom) and node.module == "app.agent.loader" for alias in node.names if alias.name in forbidden @@ -2127,7 +2129,7 @@ def test_agent_package_roots_do_not_duplicate_implementation_exports(): """Agent 与 LLM 包根不得实现动态转发，旧符号只能由精确 Compat 路由承接。""" for relative_path in ("app/agent/__init__.py", "app/agent/llm/__init__.py"): path = PROJECT_ROOT / relative_path - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + tree = _parse_source(path) assert all(isinstance(node, ast.Expr) for node in tree.body), relative_path assert set(SYMBOL_ALIASES["app.agent"]) == { @@ -2150,10 +2152,10 @@ def test_host_code_imports_agent_and_llm_symbols_from_owner_modules(): """宿主不得消费 Agent 包根兼容符号，避免包根再次成为第二公开面。""" violations: dict[str, set[str]] = {} for module_name, path in _discover_modules().items(): - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + tree = _parse_source(path) imported = { f"{node.module}.{alias.name}" - for node in ast.walk(tree) + for node in _walk_module(tree) if isinstance(node, ast.ImportFrom) and node.module in {"app.agent", "app.agent.llm"} for alias in node.names } @@ -2169,10 +2171,10 @@ def test_host_consumers_resolve_llm_provider_runtime_through_gateway(): for module_name, path in _discover_modules().items(): if module_name.startswith(("app.agent", "app.startup")): continue - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + tree = _parse_source(path) imported = { alias.name - for node in ast.walk(tree) + for node in _walk_module(tree) if isinstance(node, ast.ImportFrom) and node.module in {"app.agent.llm", "app.agent.llm.provider"} for alias in node.names if alias.name == "LLMProviderManager" @@ -2191,10 +2193,10 @@ def test_host_consumers_use_agent_audio_capability_application_port(): for module_name, path in _discover_modules().items(): if module_name.startswith(("app.agent", "app.startup")): continue - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + tree = _parse_source(path) imported = { alias.name - for node in ast.walk(tree) + for node in _walk_module(tree) if isinstance(node, ast.ImportFrom) and node.module in {"app.agent.llm", "app.agent.llm.capability"} for alias in node.names if alias.name == "AgentCapabilityManager" @@ -2246,8 +2248,8 @@ def test_modules_read_deployment_settings_through_runtime_port(): """宿主 Module 不得绕过 runtime 配置端口直接依赖 Settings 实例。""" violations: list[str] = [] for path in (APP_ROOT / "modules").rglob("*.py"): - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if not isinstance(node, ast.ImportFrom) or node.module != "app.runtime.config": continue if any(alias.name == "settings" for alias in node.names): @@ -2263,8 +2265,8 @@ def test_runtime_implementation_does_not_use_legacy_settings_proxy(): for path in (APP_ROOT / "runtime").rglob("*.py"): if path == APP_ROOT / "runtime" / "settings.py": continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) - for node in ast.walk(tree): + tree = _parse_source(path) + for node in _walk_module(tree): if not isinstance(node, ast.ImportFrom): continue if node.module != "app.runtime.settings": @@ -2295,12 +2297,12 @@ def test_deprecated_settings_proxy_imports_are_zero(): continue if path.is_relative_to(APP_ROOT / "plugins"): continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) imports_compat = any( isinstance(node, ast.ImportFrom) and node.module == "app.runtime.settings" and any(alias.name.lower().endswith("compat") for alias in node.names) - for node in ast.walk(tree) + for node in _walk_module(tree) ) if not imports_compat: continue @@ -2324,12 +2326,12 @@ def test_global_settings_imports_stay_within_compatibility_baseline(): for path in APP_ROOT.rglob("*.py"): if path.is_relative_to(APP_ROOT / "plugins"): continue - tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + tree = _parse_source(path) if any( isinstance(node, ast.ImportFrom) and node.module == "app.runtime.config" and any(alias.name == "settings" for alias in node.names) - for node in ast.walk(tree) + for node in _walk_module(tree) ): imports.add(path.relative_to(PROJECT_ROOT).as_posix()) diff --git a/tests/test_architecture_egress.py b/tests/test_architecture_egress.py index 8616d7f8b..b88ae1252 100644 --- a/tests/test_architecture_egress.py +++ b/tests/test_architecture_egress.py @@ -3,8 +3,8 @@ import json from pathlib import Path -from scripts.architecture.baseline import collect_dependency_baseline from scripts.architecture.egress import collect_direct_egress +from tests.architecture_cache import dependency_baseline PROJECT_ROOT = Path(__file__).parents[1] DEPENDENCY_POLICY_PATH = ( @@ -672,7 +672,7 @@ class Service: def test_current_egress_facts_are_complete_and_self_consistent() -> None: """当前宿主 egress identity 必须完整、可收缩、排序且统计自洽。""" - value = collect_dependency_baseline()["direct_egress"] + value = dependency_baseline()["direct_egress"] entries = value["entries"] application_entries = [ entry @@ -722,7 +722,7 @@ def test_current_egress_facts_are_complete_and_self_consistent() -> None: def test_current_egress_facts_match_exact_policy() -> None: """现存事实必须逐条分类，且 registry/scope 不得与 collector 漂移。""" - facts = collect_dependency_baseline()["direct_egress"] + facts = dependency_baseline()["direct_egress"] policy = json.loads(DEPENDENCY_POLICY_PATH.read_text(encoding="utf-8")) egress_policy = policy["direct_egress"] groups = egress_policy["groups"] diff --git a/tests/test_architecture_event_facts.py b/tests/test_architecture_event_facts.py index 97a26fe28..206fe509a 100644 --- a/tests/test_architecture_event_facts.py +++ b/tests/test_architecture_event_facts.py @@ -7,6 +7,7 @@ from scripts.architecture.event_facts import ( collect_event_facts, fingerprint_event_fact, ) +from tests.architecture_cache import current_event_facts EVENT_MEMBERS = { "EventType": ("Alpha", "Beta", "Gamma"), @@ -1135,15 +1136,7 @@ def handler(event): def test_collect_event_facts_matches_current_host_inventory() -> None: """统一事实覆盖当前 86 个 producer 调用与 17 个 consumer 调用。""" - from scripts.architecture.baseline import ( - _event_enum_members, - discover_modules, - ) - - facts = collect_event_facts( - discover_modules(), - {enum_name: _event_enum_members(enum_name) for enum_name in ("EventType", "ChainEventType")}, - ) + facts = current_event_facts() producers = facts["producers"] consumers = facts["consumers"] diff --git a/tests/test_architecture_event_policy.py b/tests/test_architecture_event_policy.py index 78e22e2ce..940f146ed 100644 --- a/tests/test_architecture_event_policy.py +++ b/tests/test_architecture_event_policy.py @@ -7,7 +7,6 @@ from typing import Any, Callable import pytest -from scripts.architecture.baseline import collect_current_event_facts from scripts.architecture.event_facts import fingerprint_event_fact from scripts.architecture.event_policy import ( DEFAULT_EVENT_POLICY_PATH, @@ -19,6 +18,7 @@ from scripts.architecture.event_policy import ( parse_args, validate_event_consumer_policy, ) +from tests.architecture_cache import current_event_facts def _fact( @@ -112,7 +112,7 @@ def test_event_policy_accepts_exact_static_and_dynamic_facts() -> None: def test_current_event_consumer_policy_matches_exact_reviewed_set() -> None: """当前 17 条宿主 consumer 必须逐条匹配人工政策且没有非法成员。""" - facts = collect_current_event_facts()["consumers"] + facts = current_event_facts()["consumers"] policy = json.loads(DEFAULT_EVENT_POLICY_PATH.read_text(encoding="utf-8")) entries = policy["event_consumers"]["entries"] diff --git a/tests/test_test_runner.py b/tests/test_test_runner.py index 43e51e917..a9cbcf17f 100644 --- a/tests/test_test_runner.py +++ b/tests/test_test_runner.py @@ -53,11 +53,11 @@ def test_split_test_files_rejects_nonpositive_shard_count() -> None: def test_recorded_durations_balance_the_current_ci_suite() -> None: - """慢文件权重必须有效，并将当前全集的八分片预计耗时维持均衡。""" + """慢文件权重必须有效，并将 CI 分片实际运行的文件集的六分片预计耗时维持均衡。""" durations = test_runner.load_test_durations() assert all(value > 0 for value in durations.values()) - test_files = test_runner.collect_test_files() - shards = test_runner.split_test_files(test_files, shard_count=8) + test_files = test_runner.collect_test_files(exclude_architecture_gate=True) + shards = test_runner.split_test_files(test_files, shard_count=6) totals = [sum(durations.get(path.name, 1.0) for path in shard) for shard in shards] assert sorted(path for shard in shards for path in shard) == test_files @@ -71,9 +71,9 @@ def test_main_defaults_to_four_parallel_shards(monkeypatch) -> None: """无 runner 参数时应并行执行四个独立 pytest 文件分片。""" test_files = _test_files(10) captured = {} - monkeypatch.setattr(test_runner, "collect_test_files", lambda: test_files) + monkeypatch.setattr(test_runner, "collect_test_files", lambda *_: test_files) - def fake_run_parallel(shards, pytest_args): + def fake_run_parallel(shards, pytest_args, _exclude_architecture_gate=False): """记录默认并行入口接收的分片与 pytest 参数。""" captured["shards"] = shards captured["pytest_args"] = pytest_args @@ -90,7 +90,7 @@ def test_main_runs_requested_ci_shard_in_current_process(monkeypatch) -> None: """CI 指定分片时只运行该分片，并继续透传 pytest 参数。""" test_files = _test_files(10) captured = {} - monkeypatch.setattr(test_runner, "collect_test_files", lambda: test_files) + monkeypatch.setattr(test_runner, "collect_test_files", lambda *_: test_files) def fake_run_pytest(paths, pytest_args): """记录 pytest 入口接收的文件与透传参数。""" @@ -122,6 +122,24 @@ def test_main_serial_preserves_legacy_full_suite_entry(monkeypatch) -> None: assert captured["pytest_args"] == ["-q", "--maxfail=1"] +def test_architecture_gate_files_are_excluded_only_on_request() -> None: + """默认全量包含门禁文件；排除后剩余文件与门禁文件恰好构成全集。""" + full = test_runner.collect_test_files() + remaining = test_runner.collect_test_files(exclude_architecture_gate=True) + gate = [path for path in full if path.name in test_runner.ARCHITECTURE_GATE_TESTS] + + assert sorted(path.name for path in gate) == sorted(test_runner.ARCHITECTURE_GATE_TESTS) + assert sorted([*remaining, *gate]) == full + + +def test_parallel_workers_inherit_architecture_gate_exclusion() -> None: + """并行 worker 必须收到同一排除选项，否则各自划分的分片会遗漏或重复文件。""" + command = test_runner._worker_command(2, 4, ["-q"], exclude_architecture_gate=True) + + assert command[2:] == ["--shard", "2/4", "--exclude-architecture-gate", "-q"] + assert "--exclude-architecture-gate" not in test_runner._worker_command(2, 4, ["-q"]) + + @pytest.mark.parametrize("value", ["0/4", "5/4", "1/0", "invalid"]) def test_invalid_shard_values_are_rejected(value: str) -> None: """分片参数必须使用有效的一基 N/TOTAL 范围。""" @@ -134,6 +152,6 @@ def test_workflow_uses_the_shared_runner_contract() -> None: workflow = WORKFLOW.read_text(encoding="utf-8") assert 'run: uv run --locked --no-sync python tests/run.py --shard' not in workflow - assert "python -m coverage run --parallel-mode tests/run.py --shard" in workflow + assert "python -m coverage run --parallel-mode tests/run.py --exclude-architecture-gate --shard" in workflow assert "mapfile" not in workflow assert "SHARD_INDEX" not in workflow