DataOps-platform/tests/test_phase3_wp00_baseline_contract.py

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 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 subprocess.run(
        ["diff", "-qr", "app", "deployment/app"], cwd=ROOT
    ).returncode == 0
    assert subprocess.run(
        ["diff", "-qr", "migrations", "deployment/migrations"], cwd=ROOT
    ).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"]
    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
    assert len(live["acceptance_evidence"]) == 8
    assert all(
        evidence["status"] == "ENGINEERING_ONLY"
        for evidence in live["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)