DataOps-platform/tests/test_phase2_wp00_baseline_contract.py

from __future__ import annotations

import json
import re
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
SCOPE_BASELINE = ROOT / "docs/phase2/P2_WP00_SCOPE_BASELINE.md"
REQUIREMENTS = ROOT / "docs/phase2/P2_WP00_REQUIREMENTS.json"
MANIFEST = ROOT / "docs/phase2/P2_WP00_BASELINE_MANIFEST.json"
FUNCTION_LEDGER = ROOT / "docs/FUNCTION_MODULE_CENSUS_20260726.md"

PHASE1_RELEASE_COMMIT = "1b926255374191b25716a947c44f121a0e327dd5"
PHASE2_START_COMMIT = "ff42feec7dda1b8bd80d0ba0f17ea31f9ffb4d2d"


def _load(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8"))


def test_wp00_deliverables_exist() -> None:
    for path in (SCOPE_BASELINE, REQUIREMENTS, MANIFEST):
        assert path.is_file(), path


def test_wp00_scope_document_freezes_the_entry_gate() -> None:
    content = SCOPE_BASELINE.read_text(encoding="utf-8")
    for heading in (
        "范围冻结",
        "第二业务域",
        "对象与样本",
        "责任与验收",
        "指标与裁剪",
        "不可变基线",
        "P2-WP01 进入门禁",
        "外部确认边界",
    ):
        assert heading in content
    assert "备品备件/物料主数据" in content
    assert "不扩展为分析开发平台" in content
    assert "OIDC SSO" in content
    assert "不能替代企业真实数据验收" in content
    assert not re.search(r"\b(?:TBD|TODO)\b|\{\{[^}]+\}\}", content)


def test_wp00_second_domain_has_actionable_samples() -> None:
    payload = _load(REQUIREMENTS)
    assert payload["work_package"] == "P2-WP00"
    domain = payload["second_business_domain"]
    assert domain["code"] == "spare_parts_material_master"
    assert domain["name"] == "备品备件/物料主数据"
    assert domain["selection_status"] == "DEFAULT_BASELINE"
    assert len(domain["governance_objects"]) >= 3
    assert len(domain["sources"]) >= 1
    assert len(domain["terms_and_code_sets"]) >= 1
    assert len(domain["quality_rule_samples"]) >= 5
    assert len(domain["workflow_samples"]) >= 1
    for collection in (
        "governance_objects",
        "sources",
        "terms_and_code_sets",
        "quality_rule_samples",
        "workflow_samples",
    ):
        assert all(item.get("id") and item.get("name") for item in domain[collection])


def test_wp00_p0_requirements_are_owned_traceable_and_accepted() -> None:
    payload = _load(REQUIREMENTS)
    requirements = payload["requirements"]
    assert {item["id"] for item in requirements} == {
        "P2-WP00-R01",
        "P2-WP00-R02",
        "P2-WP00-R03",
        "P2-WP00-R04",
        "P2-WP00-R05",
    }
    ledger = FUNCTION_LEDGER.read_text(encoding="utf-8")
    for item in requirements:
        assert item["priority"] == "P0"
        assert item["status"] == "BASELINED"
        assert item["owner_role"]
        assert item["acceptance_criterion"]
        assert item["sample_refs"]
        assert item["module_ids"]
        for module_id in item["module_ids"]:
            if re.fullmatch(r"[A-Z]{3}-\d{2}", module_id):
                assert f"| {module_id} |" in ledger


def test_wp00_metrics_and_capacity_cut_line_are_explicit() -> None:
    payload = _load(REQUIREMENTS)
    metrics = payload["acceptance_metrics"]
    assert len(metrics) >= 8
    assert all(metric["id"] and metric["definition"] and metric["target"] for metric in metrics)
    assert any(metric["id"] == "P2-M13" and metric["target"] == "0" for metric in metrics)
    cut_line = payload["capacity_cut_line"]
    assert cut_line["p0_must_keep"]
    assert cut_line["p1_reducible"]
    assert "P2-WP12" in cut_line["never_remove"]
    assert "P2-WP13" in cut_line["never_remove"]


def test_wp00_responsibility_roles_do_not_invent_enterprise_people() -> None:
    payload = _load(REQUIREMENTS)
    assignments = payload["responsibility_assignments"]
    required_roles = {
        "phase2_product_owner",
        "material_domain_owner",
        "material_data_steward",
        "phase2_technical_owner",
        "phase2_acceptance_owner",
    }
    assert required_roles.issubset({item["role"] for item in assignments})
    for item in assignments:
        assert item["accountability"]
        assert item["binding_status"] in {"PLATFORM_ASSIGNED", "ENTERPRISE_TO_ASSIGN"}
        if item["binding_status"] == "ENTERPRISE_TO_ASSIGN":
            assert item["person_name"] is None


def test_wp00_manifest_pins_source_api_migration_and_data_state() -> None:
    manifest = _load(MANIFEST)
    assert manifest["work_package"] == "P2-WP00"
    source = manifest["source_control"]
    assert source["phase1_release"]["commit"] == PHASE1_RELEASE_COMMIT
    assert source["phase1_release"]["tag"] == "v0.2.0-phase1"
    assert source["phase1_release"]["remote_branch"] == "codex/dataops-phase1-release"
    assert source["phase2_start"]["commit"] == PHASE2_START_COMMIT
    assert source["phase2_start"]["branch"] == "codex/dataops-phase2-platformization"

    api = manifest["api_contract"]
    assert api["operation_count"] == 219
    assert re.fullmatch(r"[0-9a-f]{64}", api["sha256"])
    assert api["generated_matches_committed"] is True

    migrations = manifest["migration_contract"]
    assert migrations["head"] == "20260730_360"
    assert migrations["file_count"] == 42
    assert re.fullmatch(r"[0-9a-f]{40}", migrations["git_tree"])

    release_copy = manifest["release_copy_contract"]
    assert release_copy["app_git_tree"] == release_copy["deployment_app_git_tree"]
    assert release_copy["matches"] is True

    snapshot = manifest["data_snapshot"]
    assert snapshot["scope"] == "LOCAL_ENGINEERING_BASELINE"
    assert snapshot["enterprise_uat"] is False
    assert snapshot["public_table_count"] == 116
    for table in (
        "users",
        "device_assets",
        "device_asset_source_mappings",
        "device_quality_runs",
        "device_quality_issues",
        "device_operational_events",
        "governance_audit_seals",
    ):
        assert isinstance(snapshot["record_counts"][table], int)
    assert manifest["limitations"]