DataOps-platform/tests/test_p2_wp13_acceptance_handover_contract.py

from __future__ import annotations

import json
import re
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
UAT_CASES = ROOT / "docs/acceptance/P2_WP13_UAT_CASES.json"
UAT_REPORT = ROOT / "docs/acceptance/P2_WP13_UAT_PLAN_AND_REPORT.md"
DEFECT_REGISTER = ROOT / "docs/acceptance/P2_WP13_DEFECT_REGISTER.md"
SIGNOFF = ROOT / "docs/acceptance/P2_WP13_SIGNOFF_CHECKLIST.md"
USER_GUIDE = ROOT / "docs/guides/DATAOPS_PHASE2_USER_OPERATION_GUIDE.md"
OPERATIONS_HANDOVER = ROOT / "docs/runbooks/DATAOPS_PHASE2_OPERATIONS_HANDOVER.md"
TRAINING = ROOT / "docs/training/DATAOPS_PHASE2_TRAINING_PLAN.md"
PHASE3_BACKLOG = ROOT / "docs/roadmap/DATAOPS_PHASE3_BACKLOG.md"
CENSUS = ROOT / "docs/FUNCTION_MODULE_CENSUS_20260726.md"
CENSUS_AUDIT = ROOT / "docs/validation/P2_WP13_FUNCTION_CENSUS_AUDIT.md"
EVIDENCE = ROOT / "docs/validation/P2_WP13_ACCEPTANCE_HANDOVER_EVIDENCE.md"
PHASE2_PLAN = ROOT / "docs/DATAOPS_PHASE2_3_MONTH_DEVELOPMENT_PLAN_20260730.md"

DELIVERABLES = (
    UAT_CASES,
    UAT_REPORT,
    DEFECT_REGISTER,
    SIGNOFF,
    USER_GUIDE,
    OPERATIONS_HANDOVER,
    TRAINING,
    PHASE3_BACKLOG,
    CENSUS_AUDIT,
    EVIDENCE,
)

P2_COMPLETED_LEDGER_IDS = {
    "GOV-05",
    "GOV-06",
    "GOV-07",
    "GOV-08",
    "GOV-09",
    "CAT-17",
    "CAT-23",
    "CAT-26",
    "OBS-04",
    "OBS-05",
    "OBS-06",
    "WFC-04",
    "WFC-05",
    "WFC-06",
    "WFC-07",
    "WFC-08",
    "WFC-09",
    "WFC-10",
    "WFC-11",
    "WFC-12",
    "WFC-15",
}


def _read(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def _cases() -> list[dict]:
    payload = json.loads(_read(UAT_CASES))
    assert payload["schema_version"] == "1.0"
    assert payload["work_package"] == "P2-WP13"
    return payload["cases"]


def _ledger_rows() -> dict[str, list[str]]:
    rows: dict[str, list[str]] = {}
    for line in _read(CENSUS).splitlines():
        match = re.match(r"^\| ([A-Z]{3}-\d{2}) \| (.+) \| (.+) \| (.+) \|$", line)
        if match:
            rows[match.group(1)] = [match.group(2), match.group(3), match.group(4)]
    return rows


def test_p2_wp13_deliverables_exist() -> None:
    missing = [str(path.relative_to(ROOT)) for path in DELIVERABLES if not path.is_file()]
    assert not missing, f"P2-WP13 交付物缺失: {missing}"


def test_uat_cases_cover_all_work_packages_roles_and_external_boundaries() -> None:
    cases = _cases()
    assert len(cases) >= 24
    required_fields = {
        "id",
        "work_packages",
        "module_ids",
        "title",
        "actor",
        "acceptance_level",
        "preconditions",
        "steps",
        "expected",
        "evidence",
        "result",
        "notes",
    }
    actors = {"admin", "editor", "viewer", "second_domain_owner", "enterprise_acceptance"}
    for case in cases:
        assert required_fields <= case.keys(), case.get("id")
        assert case["id"].startswith("P2-WP13-UAT-")
        assert case["actor"] in actors
        assert case["acceptance_level"] in {"LOCAL_ENGINEERING", "ENTERPRISE_FORMAL"}
        assert case["result"] in {"PASS_LOCAL", "FAIL", "BLOCKED_EXTERNAL", "NOT_RUN"}
        assert case["work_packages"] and case["module_ids"]
        assert case["preconditions"] and len(case["steps"]) >= 2
        assert case["expected"] and case["evidence"]
        if case["acceptance_level"] == "ENTERPRISE_FORMAL":
            assert case["result"] != "PASS_LOCAL"

    covered = {work_package for case in cases for work_package in case["work_packages"]}
    assert {f"P2-WP{number:02d}" for number in range(1, 13)} <= covered
    assert actors - {"enterprise_acceptance"} <= {case["actor"] for case in cases}
    assert any(case["result"] == "BLOCKED_EXTERNAL" for case in cases)
    assert not any(case["result"] == "FAIL" for case in cases)


def test_local_uat_evidence_references_existing_workspace_files() -> None:
    for case in _cases():
        if case["result"] != "PASS_LOCAL":
            continue
        for reference in case["evidence"]:
            path = ROOT / reference.split("#", 1)[0]
            assert path.is_file(), f"{case['id']} evidence missing: {reference}"


def test_acceptance_docs_are_explicit_about_local_and_enterprise_status() -> None:
    report = _read(UAT_REPORT)
    for phrase in (
        "本地工程预验收",
        "企业正式验收",
        "不能替代企业真实数据 UAT",
        "BLOCKED_EXTERNAL",
        "P0",
        "签字",
    ):
        assert phrase in report
    for path in DELIVERABLES[1:]:
        assert not re.search(r"\b(?:TBD|TODO)\b|\{\{[^}]+\}\}", _read(path)), path.name


def test_defect_register_and_signoff_are_fail_closed() -> None:
    defects = _read(DEFECT_REGISTER)
    for field in ("缺陷编号", "严重度", "状态", "复现与证据", "处置决定"):
        assert field in defects
    for severity in ("P0", "P1", "P2", "P3"):
        assert severity in defects
    assert "P0：0" in defects
    assert "未关闭 P1：0" in defects

    signoff = _read(SIGNOFF)
    for role in ("产品负责人", "业务负责人", "技术负责人", "安全负责人", "运维负责人"):
        assert role in signoff
    assert "不得预签" in signoff
    assert "BLOCKED_EXTERNAL" in signoff


def test_user_operations_and_training_materials_cover_handover_scope() -> None:
    user_guide = _read(USER_GUIDE)
    for topic in (
        "管理员",
        "编辑者",
        "查看者",
        "第二业务域责任人",
        "主动元数据",
        "语义",
        "质量",
        "事故",
        "数据产品",
        "Agent",
        "安全",
    ):
        assert topic in user_guide

    operations = _read(OPERATIONS_HANDOVER)
    for topic in (
        "每日巡检",
        "每周巡检",
        "备份",
        "恢复",
        "回滚",
        "升级",
        "异常升级",
        "责任",
        "移交清单",
    ):
        assert topic in operations

    training = _read(TRAINING)
    for topic in ("用户培训", "运维培训", "练习", "签到", "补训", "BLOCKED_EXTERNAL"):
        assert topic in training


def test_phase3_backlog_preserves_product_boundary_and_sso_gate() -> None:
    backlog = _read(PHASE3_BACKLOG)
    for field in ("待办编号", "来源模块", "优先级", "启动条件", "验收标准", "依赖", "目标阶段"):
        assert field in backlog
    for topic in ("单点登录", "OIDC", "IdP", "企业数据源", "多租户", "生产可观测", "第三业务域"):
        assert topic in backlog
    assert "不扩展为分析开发平台" in backlog


def test_function_census_has_264_unique_audited_items_and_no_stale_p2_plans() -> None:
    rows = _ledger_rows()
    assert len(rows) == 264
    assert all(row[2].strip() for row in rows.values())
    for module_id in P2_COMPLETED_LEDGER_IDS:
        assert rows[module_id][2] != "规划中", module_id

    audit = _read(CENSUS_AUDIT)
    assert "264" in audit
    assert "264/264" in audit
    assert "生产已建设" not in audit
    assert "企业 UAT" in audit


def test_phase2_plan_and_evidence_record_engineering_completion_without_formal_signoff() -> None:
    plan = _read(PHASE2_PLAN)
    section = plan.split("### P2-WP13 第二阶段验收与移交", 1)[1].split("## 8.", 1)[0]
    assert section.count("- [x]") == 5
    assert "本地工程完成" in section
    assert "企业正式验收未完成" in section

    evidence = _read(EVIDENCE)
    for section_name in (
        "文档契约",
        "UAT 覆盖",
        "角色与权限",
        "缺陷结论",
        "培训与移交",
        "功能台账",
        "外部门禁",
    ):
        assert section_name in evidence
    assert "tests/test_p2_wp13_acceptance_handover_contract.py" in evidence
    assert "PASS_LOCAL" in evidence
    assert "BLOCKED_EXTERNAL" in evidence