from __future__ import annotations import hashlib import json import re import subprocess import sys from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) from generate_p3_wp14_release_manifest import _tree_digest as _manifest_tree_digest # noqa: E402 CASES = ROOT / "docs/acceptance/P3_WP14_UAT_CASES.json" REPORT = ROOT / "docs/acceptance/P3_WP14_UAT_PLAN_AND_REPORT.md" DEFECTS = ROOT / "docs/acceptance/P3_WP14_DEFECT_AND_GATE_LEDGER.md" SIGNOFF = ROOT / "docs/acceptance/P3_WP14_FIVE_PARTY_SIGNOFF.md" RUNBOOK = ROOT / "docs/runbooks/P3_WP14_ENTERPRISE_PILOT_HANDOVER.md" EVIDENCE = ROOT / "docs/validation/P3_WP14_ENTERPRISE_PILOT_EVIDENCE.md" RELEASE = ROOT / "docs/phase3/P3_WP14_RELEASE_TRACEABILITY.json" MANIFEST = ROOT / "docs/phase3/P3_WP14_RELEASE_MANIFEST.json" AUDIT = ROOT / "docs/phase3/P3_WP14_264_MODULE_CLOSURE_AUDIT.json" CONFIG_A = ROOT / "docs/validation/P3_WP14_LOCAL_CONFIG_A.json" CONFIG_B = ROOT / "docs/validation/P3_WP14_LOCAL_CONFIG_B.json" CONFIG_COMPARISON = ROOT / "docs/validation/P3_WP14_LOCAL_CONFIG_COMPARISON.json" CONFIG_RENDER = ROOT / "docs/validation/P3_WP14_LOCAL_CONFIG_RENDER.json" VERIFICATION_LEDGER = ROOT / "docs/validation/P3_WP14_LOCAL_VERIFICATION_LEDGER.json" VERIFICATION_LOG = ROOT / "docs/validation/P3_WP14_LOCAL_VERIFICATION.log" FRESH_VERIFIER = ROOT / "scripts/verify_p3_wp14_fresh_local_engineering.py" PHASE_PLAN = ROOT / "docs/DATAOPS_PHASE3_6_MONTH_DEVELOPMENT_PLAN_20260802.md" BACKLOG = ROOT / "docs/roadmap/DATAOPS_PHASE3_BACKLOG.md" LEDGER = ROOT / "docs/FUNCTION_MODULE_CENSUS_20260726.md" REQUIREMENTS = ROOT / "docs/phase3/P3_WP00_REQUIREMENTS.json" SOURCE_HEAD = ROOT / "migrations/versions/20260818_559_plugin_three_identity_gate.py" DEPLOYMENT_HEAD = ROOT / "deployment/migrations/versions/20260818_559_plugin_three_identity_gate.py" REQUIRED_DELIVERABLES = ( CASES, REPORT, DEFECTS, SIGNOFF, RUNBOOK, EVIDENCE, RELEASE, AUDIT, MANIFEST, CONFIG_A, CONFIG_B, CONFIG_COMPARISON, CONFIG_RENDER, VERIFICATION_LEDGER, VERIFICATION_LOG, FRESH_VERIFIER, ) STARTED_WORK_PACKAGES = {f"P3-WP{number:02d}" for number in range(1, 14)} REQUIRED_ROLES = { "admin", "governance", "ordinary_user", "data_owner", "operations", "security", "audit", } REQUIRED_CATEGORIES = { "permission", "interface", "database", "browser_ui", "enterprise_boundary", "migration", "upgrade", "rollback", "restore", "target_scale", "failure", "security", "training", "operations", "five_party_signoff", "isolated_local_environment", } def _read(path: Path) -> str: return path.read_text(encoding="utf-8") def _json(path: Path) -> dict: return json.loads(_read(path)) def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _tree_digest(relative_root: str) -> str: return _manifest_tree_digest(relative_root) def _ledger_rows() -> dict[str, dict[str, str]]: content = _read(LEDGER) 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 rows[module_id] = { "module_name": module_name, "description": description, "maturity": maturity, } return rows def test_p3_wp14_deliverables_exist_and_external_boundary_is_closed() -> None: missing = [str(path.relative_to(ROOT)) for path in REQUIRED_DELIVERABLES if not path.is_file()] assert not missing, f"P3-WP14 delivery missing: {missing}" prohibited = ("PASS_ENTERPRISE", "ENTERPRISE_PASS", "ENTERPRISE_ACCEPTANCE_COMPLETE") local_artifacts = { CONFIG_A, CONFIG_B, CONFIG_COMPARISON, CONFIG_RENDER, VERIFICATION_LEDGER, VERIFICATION_LOG, MANIFEST, FRESH_VERIFIER, } for path in REQUIRED_DELIVERABLES: assert not any(token in _read(path) for token in prohibited), path assert "TBD_EXTERNAL" in _read(path) or path == AUDIT or path in local_artifacts, path def test_p3_wp14_machine_matrix_covers_started_scope_roles_and_boundaries() -> None: payload = _json(CASES) assert payload["schema_version"] == "1.0" assert payload["work_package"] == "P3-WP14" assert payload["result_definitions"]["PASS_LOCAL"] cases = payload["cases"] assert len(cases) >= 29 required_fields = { "id", "work_packages", "module_ids", "title", "actor", "categories", "acceptance_level", "preconditions", "steps", "expected", "evidence", "result", "notes", } assert len({case["id"] for case in cases}) == len(cases) for case in cases: assert required_fields <= case.keys(), case.get("id") assert case["id"].startswith("P3-WP14-UAT-") assert set(case["work_packages"]) <= STARTED_WORK_PACKAGES assert case["module_ids"] assert case["actor"] in REQUIRED_ROLES assert case["categories"] assert case["acceptance_level"] in {"LOCAL_ENGINEERING", "ENTERPRISE_FORMAL"} assert case["result"] in {"PASS_LOCAL", "FAIL", "BLOCKED_EXTERNAL", "NOT_RUN", "TBD_EXTERNAL"} assert case["preconditions"] and case["steps"] and case["expected"] and case["evidence"] if case["acceptance_level"] == "ENTERPRISE_FORMAL": assert case["result"] in {"BLOCKED_EXTERNAL", "TBD_EXTERNAL", "NOT_RUN", "FAIL"} if case["result"] == "PASS_LOCAL": for reference in case["evidence"]: assert (ROOT / reference.split("#", 1)[0]).is_file(), (case["id"], reference) assert {wp for case in cases for wp in case["work_packages"]} == STARTED_WORK_PACKAGES assert {case["actor"] for case in cases} >= REQUIRED_ROLES assert {category for case in cases for category in case["categories"]} >= REQUIRED_CATEGORIES assert {wp for case in cases if case["result"] == "PASS_LOCAL" for wp in case["work_packages"]} == STARTED_WORK_PACKAGES assert any(case["result"] == "BLOCKED_EXTERNAL" for case in cases) assert not any(case["result"] == "FAIL" for case in cases) by_id = {case["id"]: case for case in cases} assert by_id["P3-WP14-UAT-22"]["acceptance_level"] == "ENTERPRISE_FORMAL" assert by_id["P3-WP14-UAT-22"]["result"] == "BLOCKED_EXTERNAL" assert by_id["P3-WP14-UAT-30"]["acceptance_level"] == "LOCAL_ENGINEERING" assert by_id["P3-WP14-UAT-30"]["result"] == "PASS_LOCAL" def test_p3_wp14_release_traceability_is_tamper_evident_and_uses_shared_head() -> None: release = _json(RELEASE) assert release["schema_version"] == "1.0" assert release["work_package"] == "P3-WP14" assert release["release_status"] in { "IN_PROGRESS", "ENGINEERING_BASELINE_COMPLETE_ENTERPRISE_UAT_BLOCKED", } assert release["migration_head"] == "20260818_559" assert release["signature_status"] == "TBD_EXTERNAL" assert release["remote_save_status"] == "TBD_EXTERNAL" assert release["release_manifest_path"] == "docs/phase3/P3_WP14_RELEASE_MANIFEST.json" assert release["release_manifest_sha256"] == _sha256(MANIFEST) assert release["verification_ledger_sha256"] == _sha256(VERIFICATION_LEDGER) assert release["verification_log_sha256"] == _sha256(VERIFICATION_LOG) assert SOURCE_HEAD.read_bytes() == DEPLOYMENT_HEAD.read_bytes() assert 'revision = "20260818_559"' in _read(SOURCE_HEAD) assert 'down_revision = "20260818_558"' in _read(SOURCE_HEAD) manifest = _json(MANIFEST) assert manifest["schema_version"] == "1.0" assert manifest["release_identity"]["state"] == "UNCOMMITTED_WORKTREE_EVIDENCE" assert manifest["release_identity"]["signed"] is False assert manifest["release_identity"]["git_base_commit"] == subprocess.check_output( ["git", "rev-parse", "HEAD"], cwd=ROOT, text=True ).strip() assert manifest["release_identity"]["git_base_tree"] == subprocess.check_output( ["git", "rev-parse", "HEAD^{tree}"], cwd=ROOT, text=True ).strip() assert manifest["release_identity"]["worktree_clean"] is False expected_roots = {"app", "deployment/app", "migrations", "deployment/migrations"} tree_sets = { item["root"]: item for item in manifest["controlled_artifact_sets"] if item["kind"] == "tree" } assert expected_roots <= set(tree_sets) for root, item in tree_sets.items(): assert item["sha256"] == _tree_digest(root), root file_sets = { item["path"]: item["sha256"] for item in manifest["controlled_artifact_sets"] if item["kind"] == "file" } for path in ( "docs/architecture/OPENAPI.yaml", "deploy/docker/docker-compose.yml", "scripts/generate_openapi.py", "docs/acceptance/P3_WP14_UAT_CASES.json", "tests/test_phase3_wp14_acceptance_handover_contract.py", ): assert file_sets[path] == _sha256(ROOT / path) def test_p3_wp14_isolated_configuration_and_fresh_verification_ledger_are_closed() -> None: snapshots = [_json(CONFIG_A), _json(CONFIG_B)] comparison = _json(CONFIG_COMPARISON) assert comparison["schema_version"] == "1.0" assert comparison["snapshot_paths"] == [ "docs/validation/P3_WP14_LOCAL_CONFIG_A.json", "docs/validation/P3_WP14_LOCAL_CONFIG_B.json", ] assert comparison["allowed_difference_paths"] == [ "/environment/label", "/runtime/DATAOPS_LOCAL_ENV_LABEL", "/runtime/BACKEND_PORT", ] assert len({snapshot["environment"]["label"] for snapshot in snapshots}) == 2 for path, snapshot in zip((CONFIG_A, CONFIG_B), snapshots): assert set(snapshot) == {"schema_version", "environment", "runtime", "compose"} assert snapshot["schema_version"] == "1.0" assert set(snapshot["environment"]) == {"label", "is_enterprise"} assert snapshot["environment"]["is_enterprise"] is False assert set(snapshot["runtime"]) == {"DATAOPS_LOCAL_ENV_LABEL", "BACKEND_PORT"} assert set(snapshot["compose"]) == {"file", "render_command"} assert _sha256(path) == comparison["snapshot_sha256"][path.relative_to(ROOT).as_posix()] render = _json(CONFIG_RENDER) assert render["schema_version"] == "1.0" assert render["exit_code"] == 0 assert render["snapshot_sha256"] == comparison["snapshot_sha256"] assert set(render["allowed_difference_paths"]) == set(comparison["allowed_difference_paths"]) assert render["render_command"] == ["/usr/local/bin/docker-compose", "--env-file", "/dev/null", "-f", "deploy/docker/docker-compose.yml", "config", "--format", "json"] renders = render["renders"] assert set(renders) == {"LOCAL_ISOLATED_A", "LOCAL_ISOLATED_B"} service_sets = {tuple(item["services"]) for item in renders.values()} assert len(service_sets) == 1 for snapshot, label in zip(snapshots, ("LOCAL_ISOLATED_A", "LOCAL_ISOLATED_B")): item = renders[label] assert item["exit_code"] == 0 assert item["status"] == "PASS" assert item["raw_stdout_sha256"] assert item["canonical_stdout_sha256"] assert item["consumed_label"] == snapshot["runtime"]["DATAOPS_LOCAL_ENV_LABEL"] assert item["consumed_backend_port"] == snapshot["runtime"]["BACKEND_PORT"] release = _json(RELEASE) second_environment = release["second_isolated_local_environment"] assert second_environment["formal_same_signed_version_status"] == "BLOCKED_EXTERNAL" rehearsal = second_environment["local_portability_rehearsal"] assert rehearsal["result"] == "PASS_LOCAL" assert rehearsal["manifest_signed"] is False assert rehearsal["manifest_sha256"] == release["release_manifest_sha256"] assert "not the required same signed-version" in rehearsal["boundary"] ledger = _json(VERIFICATION_LEDGER) assert ledger["schema_version"] == "1.0" assert ledger["cache_class"] == "UNSIGNED_REPRODUCIBILITY_CACHE" assert ledger["pass_local_authority"] == "FRESH_VERIFIER_CURRENT_PROCESS_ONLY" assert ledger["external_signature_anchor"] == { "status": "TBD_EXTERNAL", "gate": "BLOCKED_EXTERNAL", } assert _sha256(VERIFICATION_LOG) == ledger["log_sha256"] controlled = { path: _sha256(ROOT / path) for path in ledger["controlled_input_paths"] } digest = hashlib.sha256( json.dumps(controlled, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest() assert ledger["controlled_input_sha256"] == digest pass_ids = { case["id"] for case in _json(CASES)["cases"] if case["result"] == "PASS_LOCAL" } run_case_ids = {case_id for run in ledger["runs"] for case_id in run["case_ids"]} assert run_case_ids == pass_ids for run in ledger["runs"]: assert run["command_id"] and run["command"] assert run["status"] == "PASS" assert run["exit_code"] == 0 assert run["passed_count"] > 0 assert run["run_at"].endswith("Z") assert run["stdout_sha256"] assert run["stderr_sha256"] def test_p3_wp14_release_closure_and_ledger_security_envelope_are_fail_closed() -> None: manifest = _json(MANIFEST) closure = manifest["compose_input_closure"] assert closure["compose_file"] == "deploy/docker/docker-compose.yml" assert closure["compose_sha256"] == _sha256(ROOT / closure["compose_file"]) assert closure["render_command"][-2:] == ["--format", "json"] assert closure["render_command"][1:3] == ["--env-file", "/dev/null"] assert closure["dockerignore"] == {"path": ".dockerignore", "sha256": _sha256(ROOT / ".dockerignore")} kinds = {item["kind"] for item in closure["inputs"]} assert {"build_context", "dockerfile", "requirements", "mount_input"} <= kinds for item in closure["inputs"]: path = ROOT / item["path"] assert path.exists() assert item["sha256"] == (_tree_digest(item["path"]) if path.is_dir() else _sha256(path)) images = manifest["image_reference_closure"] assert images["immutable_signed_deployability_status"] == "BLOCKED_EXTERNAL" assert images["items"] for item in images["items"]: assert item["status"] in {"IMMUTABLE_DIGEST", "UNRESOLVED_MUTABLE_TAG", "INTERNAL_BUILD_STAGE"} if item["status"] == "UNRESOLVED_MUTABLE_TAG": assert item["gate"] == "BLOCKED_EXTERNAL" ledger = _json(VERIFICATION_LEDGER) state = _json(ROOT / "docs/validation/P3_WP14_VERIFICATION_RUN_STATE.json") assert state["status"] == "COMPLETE" assert ledger["run_nonce"] == state["run_nonce"] assert ledger["expires_at"] > ledger["generated_at"] assert datetime.fromisoformat(ledger["expires_at"].replace("Z", "+00:00")) > datetime.now(timezone.utc) assert ledger["security_envelope"] == { "locked": True, "atomic_write": True, "fd_single_read": True, "minimal_environment": True, "stdout_stderr_persisted": "SANITIZED_DIGEST_ONLY", } assert _sha256(FRESH_VERIFIER) for run in ledger["runs"]: assert run["timeout_seconds"] > 0 assert run["cwd"] == "." assert run["command_sha256"] == hashlib.sha256( json.dumps(run["command"], separators=(",", ":")).encode("utf-8") ).hexdigest() assert run["script_sha256"] assert run["entry_path"] in run["dependency_sha256"] assert run["script_sha256"] == _sha256(ROOT / run["entry_path"]) assert run["dependency_sha256"] aggregate = hashlib.sha256(json.dumps(run["dependency_sha256"], sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() assert run["dependency_aggregate_sha256"] == aggregate for path, digest in run["dependency_sha256"].items(): assert _sha256(ROOT / path) == digest release = _json(RELEASE) assert release["local_verification_cache"] == { "classification": "UNSIGNED_REPRODUCIBILITY_CACHE", "pass_local_authority": "FRESH_VERIFIER_CURRENT_PROCESS_ONLY", "external_signature_status": "TBD_EXTERNAL", "enterprise_gate": "BLOCKED_EXTERNAL", } def test_p3_wp14_264_item_closure_audit_reads_source_ledger_without_promotion() -> None: audit = _json(AUDIT) rows = _ledger_rows() assert len(rows) == 264 assert audit["schema_version"] == "1.0" assert audit["source_ledger"] == "docs/FUNCTION_MODULE_CENSUS_20260726.md#12" entries = audit["module_traceability"] assert len(entries) == 264 by_id = {entry["module_id"]: entry for entry in entries} assert set(by_id) == set(rows) requirements = { item["module_id"]: item for item in _json(REQUIREMENTS)["module_traceability"] } for module_id, source in rows.items(): entry = by_id[module_id] assert entry["module_name"] == source["module_name"] assert entry["description"] == source["description"] assert entry["source_maturity"] == source["maturity"] assert entry["p3_wp00_disposition"] == requirements[module_id]["disposition"] assert entry["implementation_owner_wp"] == requirements[module_id]["implementation_owner_wp"] assert entry["closure_wp"] == requirements[module_id]["closure_wp"] assert entry["enterprise_status"] in {"TBD_EXTERNAL", "BLOCKED_EXTERNAL", "NOT_APPLICABLE"} assert entry["evidence_refs"] def test_p3_wp14_report_ledger_signoff_runbook_and_status_sync_are_fail_closed() -> None: report = _read(REPORT) for phrase in ( "本地工程验收", "企业正式 UAT", "不能替代", "P0", "P1", "TBD_EXTERNAL", "BLOCKED_EXTERNAL", "20260818_559", ): assert phrase in report defects = _read(DEFECTS) for phrase in ("P0：0", "未关闭 P1：0", "正式决定", "TBD_EXTERNAL", "BLOCKED_EXTERNAL"): assert phrase in defects signoff = _read(SIGNOFF) for role in ("产品", "业务", "技术", "安全", "运维"): assert role in signoff assert "不得预签" in signoff runbook = _read(RUNBOOK) for phrase in ("升级", "回滚", "恢复", "每日巡检", "每周巡检", "培训", "远端保存建议"): assert phrase in runbook plan = _read(PHASE_PLAN) section = plan.split("### P3-WP14 企业试点验收、移交与台账更新", 1)[1].split("## 8.", 1)[0] assert ( "ENGINEERING_BASELINE_COMPLETE_ENTERPRISE_UAT_BLOCKED" in section or "IN_PROGRESS" in section ) assert "企业正式 UAT" in section assert "TBD_EXTERNAL" in section backlog = _read(BACKLOG) assert "P3-WP14" in backlog assert "P3_WP14_264_MODULE_CLOSURE_AUDIT.json" in backlog