from __future__ import annotations import hashlib import io import json import os import re import subprocess import tarfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SCOPE = ROOT / "docs/phase3/P3_WP00_SCOPE_BASELINE.md" REQUIREMENTS = ROOT / "docs/phase3/P3_WP00_REQUIREMENTS.json" MANIFEST = ROOT / "docs/phase3/P3_WP00_BASELINE_MANIFEST.json" REGISTERS = ROOT / "docs/phase3/P3_WP00_EXECUTION_REGISTERS.json" LEDGER = ROOT / "docs/FUNCTION_MODULE_CENSUS_20260726.md" PHASE3_PLAN = ROOT / "docs/DATAOPS_PHASE3_6_MONTH_DEVELOPMENT_PLAN_20260802.md" PHASE2_COMMIT = "95023e59bfb9f00be4ed61f7a8307fac56fbcee9" PHASE3_CAPTURE_COMMIT = "487b57d538f619407f0f2be1db7061b055b13400" P0_WORK_PACKAGES = { "P3-WP00", "P3-WP01", "P3-WP02", "P3-WP03", "P3-WP04", "P3-WP05", "P3-WP06", "P3-WP08", "P3-WP14", } def _load(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8")) def _ledger_rows() -> dict[str, dict[str, str]]: content = LEDGER.read_text(encoding="utf-8") section = content[content.index("## 12. 企业级完整产品功能模块台账") :] rows: dict[str, dict[str, str]] = {} for line in section.splitlines(): cells = [cell.strip() for cell in line.strip().strip("|").split("|")] if len(cells) == 4 and re.fullmatch(r"[A-Z]{3}-\d{2}", cells[0]): module_id, module_name, description, maturity = cells assert module_id not in rows rows[module_id] = { "module_name": module_name, "description": description, "maturity_source_text": maturity, } return rows def _git(*args: str) -> str: return subprocess.check_output(["git", *args], cwd=ROOT, text=True).strip() def _git_bytes(*args: str) -> bytes: return subprocess.check_output(["git", *args], cwd=ROOT) def _normalized_maturity(source: str) -> tuple[str, str]: mappings = ( ("已建设", "BUILT", "已建设"), ("工程完成", "ENGINEERING_COMPLETE", "工程完成"), ("部分建设", "PARTIAL", "部分建设"), ("规划中", "PLANNED", "规划中"), ("能力预留", "RESERVED", "能力预留"), ("明确不扩展", "PRODUCT_BOUNDARY", "明确不扩展"), ("兼容保留/退役中", "RETIREMENT", "兼容保留/退役中"), ("影子模式", "SHADOW", "影子模式"), ) for prefix, normalized, label in mappings: if source.startswith(prefix): return normalized, label raise AssertionError(f"unknown maturity: {source}") def _operation_count(content: str) -> int: return len( re.findall( r"^ (?:get|post|put|patch|delete|options|head):$", content, flags=re.MULTILINE, ) ) def _requirement_engineering_statuses(requirements: dict) -> set[str]: """Read only the documented WP status fields, never arbitrary nested strings.""" assert requirements["schema_version"] == "1.1" packages = [*requirements["p0_work_packages"], *requirements["non_p0_work_packages"]] expected = {f"P3-WP{number:02d}" for number in range(15)} assert {package["work_package"] for package in packages} == expected assert len(packages) == len(expected) statuses = {requirements["baseline_status"]} for package in packages: assert isinstance(package["work_package"], str) status = package.get("engineering_completion_status", package.get("engineering_status")) assert isinstance(status, str), package["work_package"] statuses.add(status) return {status for status in statuses if status.startswith("ENGINEERING_")} def _release_tree_diff(left: str, right: str) -> subprocess.CompletedProcess[str]: """Compare release source trees without treating interpreter caches as release inputs.""" return subprocess.run( ["diff", "-qr", "--exclude=__pycache__", "--exclude=*.pyc", left, right], cwd=ROOT, text=True, ) def test_wp00_deliverables_exist_and_scope_is_honest() -> None: for path in (SCOPE, REQUIREMENTS, MANIFEST, REGISTERS): assert path.is_file(), path content = SCOPE.read_text(encoding="utf-8") for heading in ( "范围冻结", "264 项模块追踪", "企业输入", "责任与退出条件", "验收指标", "容量与排期", "外部门禁", ): assert heading in content assert "工程基线完成待企业输入确认" in content assert "不代表企业\nUAT" in content or "不代表企业 UAT" in content assert "不得用 Mock、本地测试或本地数据库计数替代企业验收" in content plan = PHASE3_PLAN.read_text(encoding="utf-8") assert "TBD_AFTER_COMMIT" not in plan assert "git log -1 -- docs/phase3/P3_WP00_SCOPE_BASELINE.md" in plan def test_all_264_ledger_modules_are_traced_exactly_once() -> None: ledger = _ledger_rows() assert len(ledger) == 264 traceability = _load(REQUIREMENTS)["module_traceability"] assert len(traceability) == 264 traced = {item["module_id"]: item for item in traceability} assert len(traced) == 264 assert set(traced) == set(ledger) allowed = { "P3_WORK_PACKAGE", "FUTURE_PHASE", "MAINTAIN", "PRODUCT_BOUNDARY", "RETIREMENT", } for module_id, ledger_item in ledger.items(): item = traced[module_id] assert item["module_name"] == ledger_item["module_name"] assert item["description"] == ledger_item["description"] assert item["maturity_source_text"] == ledger_item["maturity_source_text"] normalized, current_label = _normalized_maturity(ledger_item["maturity_source_text"]) assert item["normalized_maturity"] == normalized assert item["current_maturity"] == current_label assert item["disposition"] in allowed assert isinstance(item["target_work_packages"], list) assert item["rationale"].strip() if item["disposition"] == "P3_WORK_PACKAGE": assert item["target_work_packages"] assert item["implementation_owner_wp"] assert item["closure_wp"] == "P3-WP14" assert item["implementation_owner_wp"] in item["target_work_packages"] assert set(item["dependent_wps"]).issubset(item["target_work_packages"]) else: assert item["implementation_owner_wp"] is None assert item["closure_wp"] is None def test_all_original_backlogs_are_mapped_and_splits_are_recorded() -> None: payload = _load(REQUIREMENTS) traceability = {item["module_id"]: item for item in payload["module_traceability"]} valid_module_ids = set(traceability) valid_work_packages = {f"P3-WP{number:02d}" for number in range(0, 15)} mappings = payload["backlog_work_package_mapping"] assert {item["original_backlog_id"] for item in mappings} == { f"P3-{number:02d}" for number in range(1, 13) } assert all(item["target_work_packages"] for item in mappings) declared_role_pairs: set[tuple[str, str]] = set() for mapping in mappings: assert set(mapping["module_ids"]).issubset(valid_module_ids) assert set(mapping["target_work_packages"]).issubset(valid_work_packages) for module_id in mapping["module_ids"]: traced = traceability[module_id] roles = {traced["implementation_owner_wp"], *traced["dependent_wps"]} matching_roles = roles.intersection(mapping["target_work_packages"]) assert matching_roles, (mapping["original_backlog_id"], module_id, roles) declared_role_pairs.update((module_id, role) for role in matching_roles) changes = payload["scope_additions"] assert changes for item in changes: assert item["original_backlog_id"] in {f"P3-{number:02d}" for number in range(1, 13)} assert item["added_module_ids"] assert set(item["added_module_ids"]).issubset(valid_module_ids) assert item["split_work_packages"] assert set(item["split_work_packages"]).issubset(valid_work_packages) assert item["change_reason"] addition_map = { mapping["module_id"]: mapping for mapping in item["added_module_work_package_map"] } assert set(addition_map) == set(item["added_module_ids"]) for module_id, mapping in addition_map.items(): traced = traceability[module_id] assert mapping["implementation_owner_wp"] == traced["implementation_owner_wp"] assert mapping["dependent_wps"] == traced["dependent_wps"] assert mapping["closure_wp"] == traced["closure_wp"] == "P3-WP14" declared_role_pairs.add((module_id, mapping["implementation_owner_wp"])) declared_role_pairs.update((module_id, wp) for wp in mapping["dependent_wps"]) traced_role_pairs = { (module_id, wp) for module_id, item in traceability.items() if item["disposition"] == "P3_WORK_PACKAGE" for wp in [item["implementation_owner_wp"], *item["dependent_wps"]] } assert declared_role_pairs == traced_role_pairs assert all( item["closure_wp"] == "P3-WP14" for item in traceability.values() if item["disposition"] == "P3_WORK_PACKAGE" ) def test_p0_work_packages_have_complete_accountability_and_exit_contracts() -> None: packages = _load(REQUIREMENTS)["p0_work_packages"] assert {item["work_package"] for item in packages} == P0_WORK_PACKAGES for item in packages: assert item["owner_role"] assert item["required_inputs"] assert item["acceptance_owner_role"] assert item["exit_criteria"] assert item["owner_person_name"] is None assert item["acceptance_owner_person_name"] is None assert item["enterprise_gate_inputs"] == item["required_inputs"] assert item["enterprise_confirmation_status"] == "TBD_EXTERNAL" wp00 = next(item for item in packages if item["work_package"] == "P3-WP00") assert wp00["engineering_completion_status"] == "COMPLETE" assert wp00["engineering_baseline_inputs"] wp14 = next(item for item in packages if item["work_package"] == "P3-WP14") assert any("P0 不允许" in criterion for criterion in wp14["exit_criteria"]) assert set(wp14["dynamic_required_inputs_by_started_wp"]) == { "P3-WP09", "P3-WP10", "P3-WP11", "P3-WP12", "P3-WP13" } def test_enterprise_inputs_remain_external_and_block_the_right_packages() -> None: payload = _load(REQUIREMENTS) inputs = payload["enterprise_inputs"] required_ids = { "pilot_goal", "identity", "enterprise_source_1", "enterprise_source_2", "network", "monitoring", "smtp", "collaboration", "itsm", "security_legal", "infrastructure", "third_business_domain", "ai", "multi_tenancy", "bi_ai", "cost", "plugin", } assert {item["id"] for item in inputs} == required_ids for item in inputs: assert item["status"] == "TBD_EXTERNAL" assert item["owner_role"] assert item["required_by_wp"] assert item["evidence"] == "TBD_EXTERNAL" assert item["blocking_effect"] assert item["enterprise_name"] is None assert item["due_week"] >= 1 assert item["decision_role"] assert item["evidence_ref"] == "TBD_EXTERNAL" assert item["waivable"] is False by_input = {item["id"]: item for item in inputs} package_inputs: dict[str, set[str]] = {} for package in payload["p0_work_packages"]: package_inputs.setdefault(package["work_package"], set()).update(package["required_inputs"]) for input_id in package["required_inputs"]: assert package["work_package"] in by_input[input_id]["required_by_wp"] for started_wp, input_ids in package["dynamic_required_inputs_by_started_wp"].items(): package_inputs.setdefault(started_wp, set()).update(input_ids) package_inputs.setdefault("P3-WP14", set()).update(input_ids) for input_id in input_ids: assert started_wp in by_input[input_id]["required_by_wp"] assert "P3-WP14" in by_input[input_id]["required_by_wp"] for package in payload["non_p0_work_packages"]: package_inputs.setdefault(package["work_package"], set()).update(package["required_inputs"]) for input_id in package["required_inputs"]: assert package["work_package"] in by_input[input_id]["required_by_wp"] for input_id, item in by_input.items(): for work_package in item["required_by_wp"]: assert input_id in package_inputs[work_package] assert "P3-WP07" in by_input["infrastructure"]["required_by_wp"] def test_exactly_19_default_acceptance_metrics_are_frozen_without_fake_values() -> None: metrics = _load(REQUIREMENTS)["acceptance_metrics"] assert len(metrics) == 19 assert len({metric["id"] for metric in metrics}) == 19 for metric in metrics: for key in ( "definition", "target", "denominator_or_sample", "window", "source", "approver_role", "status", ): assert metric[key] not in (None, "", []) assert metric["status"] in {"DEFAULT_BASELINED", "TBD_EXTERNAL"} enterprise_overrides = _load(REQUIREMENTS)["enterprise_metric_overrides"] assert enterprise_overrides["status"] == "TBD_EXTERNAL" assert enterprise_overrides["enterprise_name"] is None assert enterprise_overrides["records"] == [] assert set(enterprise_overrides["record_schema"]) == { "metric_id", "old_target", "new_target", "direction", "reason", "approved_at", "approver_role", "approver_person_name", "evidence_ref", } def test_manifest_replays_historical_api_migration_and_release_facts( tmp_path: Path, ) -> None: manifest = _load(MANIFEST) source = manifest["source_control"] assert source["captured_local_commit"] == PHASE3_CAPTURE_COMMIT captured_commit = source["captured_local_commit"] assert source["phase2_baseline_tag"] == "v0.3.0-phase2" assert source["phase2_baseline_commit"] == PHASE2_COMMIT assert source["merge_base_with_phase2"] == PHASE2_COMMIT assert source["remote_phase2_tracking_commit"] == PHASE2_COMMIT subprocess.run( ["git", "cat-file", "-e", f"{captured_commit}^{{commit}}"], cwd=ROOT, check=True, ) assert _git("rev-parse", "v0.3.0-phase2^{commit}") == PHASE2_COMMIT assert _git("rev-parse", f"{captured_commit}^") == PHASE2_COMMIT assert _git("merge-base", captured_commit, "v0.3.0-phase2") == PHASE2_COMMIT assert source["remote_network_refresh_performed"] is False assert "remote-tracking" in source["notes"] archive_bytes = _git_bytes( "archive", captured_commit, "app", "alembic.ini", "docs/architecture/OPENAPI.yaml", "migrations", "scripts/generate_openapi.py", ) archive_root = tmp_path / "captured" archive_root.mkdir() with tarfile.open(fileobj=io.BytesIO(archive_bytes), mode="r:") as archive: archive.extractall(archive_root) api = manifest["api_contract"] captured_openapi = _git_bytes( "show", f"{captured_commit}:docs/architecture/OPENAPI.yaml" ) assert captured_openapi == (archive_root / "docs/architecture/OPENAPI.yaml").read_bytes() generated_openapi = archive_root / "generated-openapi.yaml" generated = subprocess.run( [ str(ROOT / ".venv/bin/python"), str(archive_root / "scripts/generate_openapi.py"), "--output", str(generated_openapi), ], cwd=archive_root, env={**os.environ, "PYTHONPATH": str(archive_root)}, text=True, capture_output=True, check=True, ) generated_bytes = generated_openapi.read_bytes() actual_operation_count = _operation_count(captured_openapi.decode("utf-8")) assert str(actual_operation_count) in generated.stdout assert api["operation_count"] == actual_operation_count assert _operation_count(generated_bytes.decode("utf-8")) == actual_operation_count assert api["sha256"] == hashlib.sha256(captured_openapi).hexdigest() assert hashlib.sha256(generated_bytes).hexdigest() == api["sha256"] assert api["generated_matches_committed"] is True migration = manifest["migration_contract"] heads_output = subprocess.check_output( [ str(ROOT / ".venv/bin/alembic"), "-c", str(archive_root / "alembic.ini"), "heads", ], cwd=archive_root, env={**os.environ, "PYTHONPATH": str(archive_root)}, text=True, ) actual_heads = re.findall(r"^(\S+) \(head\)$", heads_output, flags=re.MULTILINE) assert migration["heads"] == actual_heads assert migration["file_count"] == len( list((archive_root / "migrations/versions").glob("*.py")) ) assert migration["git_tree"] == _git( "rev-parse", f"{captured_commit}:migrations" ) release = manifest["release_copy_contract"] assert release["app_git_tree"] == _git( "rev-parse", f"{captured_commit}:app" ) assert release["deployment_app_git_tree"] == _git( "rev-parse", f"{captured_commit}:deployment/app" ) assert release["migrations_git_tree"] == _git( "rev-parse", f"{captured_commit}:migrations" ) assert release["deployment_migrations_git_tree"] == _git( "rev-parse", f"{captured_commit}:deployment/migrations" ) assert release["app_git_tree"] == release["deployment_app_git_tree"] assert release["migrations_git_tree"] == release["deployment_migrations_git_tree"] def test_current_release_copies_are_recursively_in_sync() -> None: assert _release_tree_diff("app", "deployment/app").returncode == 0 assert _release_tree_diff("migrations", "deployment/migrations").returncode == 0 def test_historical_manifest_does_not_pin_future_checkout_growth() -> None: manifest = _load(MANIFEST) captured_commit = manifest["source_control"]["captured_local_commit"] captured_openapi = _git_bytes( "show", f"{captured_commit}:docs/architecture/OPENAPI.yaml" ).decode("utf-8") historical_count = _operation_count(captured_openapi) hypothetical_future_openapi = ( captured_openapi + "\n /future-baseline-proof:\n" + " get:\n" + " operationId: futureBaselineProof\n" ) assert _operation_count(hypothetical_future_openapi) == historical_count + 1 assert manifest["api_contract"]["operation_count"] == historical_count historical_heads = manifest["migration_contract"]["heads"] hypothetical_future_heads = [*historical_heads, "20990101_future"] assert len(hypothetical_future_heads) == len(historical_heads) + 1 assert manifest["migration_contract"]["heads"] == historical_heads def test_database_unavailability_is_explicit_not_fabricated() -> None: manifest = _load(MANIFEST) snapshot = manifest["data_snapshot"] assert snapshot["read_only_capture"] is True assert snapshot["enterprise_uat"] is False assert snapshot["status"] in {"AVAILABLE", "NOT_AVAILABLE"} if snapshot["status"] == "NOT_AVAILABLE": assert snapshot["error_category"] assert snapshot["alembic_current"] == "NOT_AVAILABLE" assert snapshot["public_table_count"] == "NOT_AVAILABLE" assert snapshot["record_counts"] == "NOT_AVAILABLE" results = manifest["verification_results"] assert {item["id"] for item in results} == {f"P3-EVD-{number:03d}" for number in range(1, 9)} assert all(isinstance(item["exit_code"], int) for item in results) assert all(item["enterprise_uat"] is False for item in results) current = next(item for item in results if item["id"] == "P3-EVD-005") assert current["result"] == "NOT_AVAILABLE" assert current["exit_code"] == 1 def test_capacity_schedule_change_rules_and_register_templates_are_complete() -> None: registers = _load(REGISTERS) assert registers["capacity_decision"]["status"] == "TBD_EXTERNAL" assert registers["capacity_decision"]["schedule_is_commitment"] is False assert registers["capacity_decision"]["profile_selection_mode"] == "EXACTLY_ONE" assert registers["capacity_decision"]["profiles_mutually_exclusive"] is True assert set(registers["capacity_decision"]["profiles"]) == { "PROFILE_1_2", "PROFILE_3", "PROFILE_4_5" } defined_work_packages = {f"P3-WP{number:02d}" for number in range(0, 15)} profiles = registers["capacity_decision"]["profiles"] for profile in profiles.values(): committed = set(profile["committed_wps"]) eligible = set(profile["eligible_non_p0_wps"]) deferred = set(profile["deferred_wps"]) assert P0_WORK_PACKAGES.issubset(committed) assert committed | eligible | deferred == defined_work_packages assert committed.isdisjoint(eligible) assert committed.isdisjoint(deferred) assert eligible.isdisjoint(deferred) assert all( work_package in defined_work_packages for work_package in committed | eligible | deferred ) assert "P3-WP09" in profiles["PROFILE_1_2"]["eligible_non_p0_wps"] assert ( profiles["PROFILE_1_2"]["delivery_depth"]["P3-WP09"] == "模型网关/预算/沙箱最小闭环" ) assert registers["weekly_schedule_contract"]["status"] == "UNAPPROVED_REFERENCE_SEQUENCE" assert registers["weekly_schedule_contract"]["is_delivery_commitment"] is False assert len(registers["weekly_schedule"]) == 24 assert all(item["approval_status"] == "TBD_EXTERNAL" for item in registers["weekly_schedule"]) assert registers["scope_change_rules"] templates = registers["templates"] acceptance_statuses = set(templates["acceptance_evidence"]["status_values"]) requirement_statuses = _requirement_engineering_statuses(_load(REQUIREMENTS)) assert requirement_statuses assert requirement_statuses <= acceptance_statuses assert {"requirement", "risk", "external_gate", "defect", "acceptance_evidence"} == set(templates) live = registers["registers"] collection_names = { "requirement": "requirements", "risk": "risks", "external_gate": "external_gates", "defect": "defects", "acceptance_evidence": "acceptance_evidence", } all_ids: list[str] = [] for template_name, collection_name in collection_names.items(): template = templates[template_name] records = live[collection_name] assert template["required_fields"] assert template["status_values"] assert records id_regex = re.compile( "^" + re.escape(template["id_pattern"]).replace(r"\#", r"\d") + "$" ) for record in records: assert set(template["required_fields"]).issubset(record) assert id_regex.fullmatch(record["id"]) assert record["status"] in template["status_values"] all_ids.append(record["id"]) assert len(all_ids) == len(set(all_ids)) assert len(live["external_gates"]) == 17 acceptance_evidence = live["acceptance_evidence"] assert len(acceptance_evidence) >= 8 assert all( evidence["status"].startswith("ENGINEERING_") for evidence in acceptance_evidence ) required_gate_fields = { "due_week", "decision_role", "evidence_ref", "waivable", "waiver_policy" } for gate in live["external_gates"]: assert required_gate_fields.issubset(gate) assert gate["waivable"] is False requirements_by_input = { item["id"]: item["required_by_wp"] for item in _load(REQUIREMENTS)["enterprise_inputs"] } assert { gate["input_id"]: gate["required_by_wp"] for gate in live["external_gates"] } == requirements_by_input assert "P0_SECURITY" in registers["non_waivable_gate_classes"] def test_untracked_inventory_records_paths_only_and_protects_user_files() -> None: inventory = _load(MANIFEST)["user_untracked_inventory"] assert inventory["inspection_mode"] == "PATHS_ONLY_CONTENT_NOT_READ" assert inventory["modified"] is False paths = inventory["paths"] assert paths assert all(path.startswith("canvas/") or path.startswith("docs/architecture/") for path in paths) assert any(path.startswith("canvas/") for path in paths) assert any(path.startswith("docs/architecture/") for path in paths)