DataOps-platform/tests/test_phase3_wp14_acceptance_handover_contract.py

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