DataOps-platform/tests/test_phase3_wp01_enterprise_acceptance_contract.py

from __future__ import annotations

import copy
import hashlib
import importlib.util
import json
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
TEMPLATE = ROOT / "docs/acceptance/P3_WP01_ENTERPRISE_ACCEPTANCE.json"
VALIDATOR_PATH = ROOT / "scripts/validate_phase3_wp01_acceptance.py"

SPEC = importlib.util.spec_from_file_location("phase3_wp01_validator", VALIDATOR_PATH)
assert SPEC and SPEC.loader
validator = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(validator)


def _load_template() -> dict:
    return json.loads(TEMPLATE.read_text(encoding="utf-8"))


def _bind_evidence(value: object, root: Path, counter: list[int]) -> None:
    if isinstance(value, dict):
        for key, child in value.items():
            if key == "evidence_refs" and isinstance(child, list):
                for reference in child:
                    counter[0] += 1
                    name = f"evidence-{counter[0]:03d}.json"
                    content = json.dumps(
                        {"evidence_id": counter[0], "enterprise_formal": True},
                        sort_keys=True,
                    ).encode()
                    (root / name).write_bytes(content)
                    reference.update(
                        {
                            "path": name,
                            "sha256": hashlib.sha256(content).hexdigest(),
                            "evidence_level": "ENTERPRISE_FORMAL",
                        }
                    )
            else:
                _bind_evidence(child, root, counter)
    elif isinstance(value, list):
        for child in value:
            _bind_evidence(child, root, counter)


def _accepted_bundle(tmp_path: Path) -> Path:
    payload = copy.deepcopy(_load_template())
    payload["declared_status"] = "ENTERPRISE_ACCEPTANCE_COMPLETE"
    payload["execution_context"].update(
        {
            "enterprise_name": "Synthetic Enterprise",
            "environment_id": "preprod-synthetic-01",
            "environment_type": "PREPRODUCTION",
            "executed_commit": "95023e59bfb9f00be4ed61f7a8307fac56fbcee9",
            "started_at": "2026-08-03T09:00:00+08:00",
            "completed_at": "2026-08-05T18:00:00+08:00",
        }
    )
    payload["source"].update(
        {
            "source_id": "enterprise-source-synthetic-01",
            "source_type": "POSTGRESQL",
            "readonly": True,
            "credentials_in_package": False,
            "network_zone": "enterprise-preproduction",
        }
    )
    payload["sample_summary"].update(
        {
            "classification": "DESENSITIZED_ENTERPRISE_SAMPLE",
            "object_count": 100,
            "row_count": 1000,
            "sha256": hashlib.sha256(b"synthetic-enterprise-sample-summary").hexdigest(),
            "raw_data_in_evidence": False,
        }
    )
    payload["release_artifact"].update(
        {
            "version": "v0.3.0-phase2",
            "artifact_name": "dataops-phase2-synthetic.tar.gz",
            "sha256": hashlib.sha256(b"synthetic-release-artifact").hexdigest(),
        }
    )

    for index, case in enumerate(payload["uat_cases"], start=1):
        case.update(
            {
                "result": "ENTERPRISE_PASS",
                "environment_id": "preprod-synthetic-01",
                "executed_at": f"2026-08-03T1{index}:00:00+08:00",
                "executed_by_person_id": f"executor-{index}",
                "approved_by_person_id": f"approver-{index}",
            }
        )

    metric_values = {
        "core_object_onboarding_rate": (95, 100),
        "incremental_change_accuracy": (99, 100),
        "responsibility_coverage": (95, 100),
        "quality_issue_closure_rate": (85, 100),
        "incident_evidence_coverage": (10, 10),
        "data_product_evidence_coverage": (10, 10),
    }
    for metric in payload["metrics"]:
        numerator, denominator = metric_values[metric["id"]]
        metric.update(
            {
                "status": "ENTERPRISE_PASS",
                "numerator": numerator,
                "denominator": denominator,
                "value": numerator / denominator * 100,
                "window_start": "2026-08-03T09:00:00+08:00",
                "window_end": "2026-08-03T18:00:00+08:00",
                "approved_by_person_id": f"metric-approver-{metric['id']}",
            }
        )

    rehearsal = payload["preproduction_rehearsal"]
    rehearsal.update(
        {
            "status": "ENTERPRISE_PASS",
            "environment_id": "preprod-synthetic-01",
            "executed_at": "2026-08-03T14:00:00+08:00",
            "executed_by_person_id": "preprod-operator-1",
            "approved_by_person_id": "preprod-approver-1",
            "rto_minutes": 25,
            "rpo_minutes": 10,
        }
    )
    for step in rehearsal["steps"]:
        step["status"] = "ENTERPRISE_PASS"

    for index, record in enumerate(payload["training_records"], start=1):
        record.update(
            {
                "status": "ENTERPRISE_PASS",
                "participants": [f"trainee-{index}-1", f"trainee-{index}-2"],
                "trainer_person_id": f"trainer-{index}",
                "completed_at": f"2026-08-04T1{index}:00:00+08:00",
                "exercise_passed": True,
                "retraining_required": False,
                "retraining_completed": False,
            }
        )

    payload["defect_gate"].update(
        {
            "status": "ENTERPRISE_PASS",
            "p0_open": 0,
            "p1_open": 0,
            "reviewed_by_business_person_id": "business-defect-reviewer",
            "reviewed_by_technical_person_id": "technical-defect-reviewer",
            "reviewed_at": "2026-08-04T15:00:00+08:00",
        }
    )
    payload["defect_gate"]["p1_disposition"].update(
        {"status": "NOT_REQUIRED", "covered_p1_count": 0}
    )

    for index, signoff in enumerate(payload["signoffs"], start=1):
        signoff.update(
            {
                "status": "SIGNED",
                "person_id": f"signer-{index}",
                "person_name": f"Synthetic Signer {index}",
                "signed_at": f"2026-08-05T1{index}:00:00+08:00",
            }
        )

    _bind_evidence(payload, tmp_path, [0])
    package = tmp_path / "enterprise-acceptance.json"
    package.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    return package


def _replace_first_case_evidence(package: Path, content: bytes) -> None:
    payload = json.loads(package.read_text())
    reference = payload["uat_cases"][0]["evidence_refs"][0]
    (package.parent / reference["path"]).write_bytes(content)
    reference["sha256"] = hashlib.sha256(content).hexdigest()
    package.write_text(json.dumps(payload), encoding="utf-8")


def _set_nested(payload: object, path: tuple[object, ...], value: object) -> None:
    current = payload
    for part in path[:-1]:
        current = current[part]  # type: ignore[index]
    current[path[-1]] = value  # type: ignore[index]


def test_checked_in_template_is_engineering_ready_and_blocked_external() -> None:
    result = validator.validate_acceptance_bundle(TEMPLATE)
    assert result["status"] == "BLOCKED"
    assert result["declared_status"] == "ENGINEERING_READY_BLOCKED_EXTERNAL"
    assert result["blockers"]
    assert not result["failures"]


def test_complete_synthetic_enterprise_bundle_is_accepted(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    result = validator.validate_acceptance_bundle(package)
    assert result == {
        "status": "ACCEPTED",
        "declared_status": "ENTERPRISE_ACCEPTANCE_COMPLETE",
        "blockers": [],
        "failures": [],
    }


def test_low_metric_and_local_pass_cannot_be_accepted(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload["metrics"][0]["numerator"] = 94
    payload["metrics"][0]["value"] = 94
    payload["uat_cases"][0]["result"] = "PASS_LOCAL"
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("threshold" in failure for failure in result["failures"])
    assert any("PASS_LOCAL" in failure for failure in result["failures"])


@pytest.mark.parametrize("invalid_path", ["../escape.json", "/tmp/absolute.json"])
def test_evidence_path_escape_is_rejected(tmp_path: Path, invalid_path: str) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload["uat_cases"][0]["evidence_refs"][0]["path"] = invalid_path
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("evidence path" in failure for failure in result["failures"])


def test_missing_or_digest_mismatched_evidence_is_rejected(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    first = payload["uat_cases"][0]["evidence_refs"][0]
    (tmp_path / first["path"]).unlink()
    payload["uat_cases"][1]["evidence_refs"][0]["sha256"] = "0" * 64
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("missing evidence" in failure for failure in result["failures"])
    assert any("digest mismatch" in failure for failure in result["failures"])


def test_placeholder_file_cannot_pretend_to_be_enterprise_evidence(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    reference = payload["uat_cases"][0]["evidence_refs"][0]
    content = b"TBD_EXTERNAL"
    (tmp_path / reference["path"]).write_bytes(content)
    reference["sha256"] = hashlib.sha256(content).hexdigest()
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("placeholder evidence" in failure for failure in result["failures"])


@pytest.mark.parametrize(
    "content",
    [
        b"status: TBD_EXTERNAL\nThe execution steps are documented below.",
        json.dumps({"status": "TBD_EXTERNAL", "enterprise_formal": True}).encode(),
        b"| field | value |\n|---|---|\n| result | TBD_EXTERNAL |",
        b"field,value\nresult,BLOCKED_EXTERNAL",
    ],
)
def test_structured_unfinished_state_cannot_be_formal_evidence(
    tmp_path: Path, content: bytes
) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    reference = payload["uat_cases"][0]["evidence_refs"][0]
    (tmp_path / reference["path"]).write_bytes(content)
    reference["sha256"] = hashlib.sha256(content).hexdigest()
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("placeholder evidence" in failure for failure in result["failures"])


def test_checked_in_execution_guide_cannot_be_disguised_as_formal_evidence(
    tmp_path: Path,
) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    reference = payload["uat_cases"][0]["evidence_refs"][0]
    content = (ROOT / "docs/acceptance/P3_WP01_EXECUTION_GUIDE.md").read_bytes()
    (tmp_path / reference["path"]).write_bytes(content)
    reference["sha256"] = hashlib.sha256(content).hexdigest()
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("placeholder evidence" in failure for failure in result["failures"])


def test_formal_evidence_may_describe_historical_local_or_blocked_results(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    reference = payload["uat_cases"][0]["evidence_refs"][0]
    content = b"Historical PASS_LOCAL and BLOCKED_EXTERNAL results were superseded by this formal run."
    (tmp_path / reference["path"]).write_bytes(content)
    reference["sha256"] = hashlib.sha256(content).hexdigest()
    package.write_text(json.dumps(payload), encoding="utf-8")
    assert validator.validate_acceptance_bundle(package)["status"] == "ACCEPTED"


@pytest.mark.parametrize(
    "content",
    [
        json.dumps(
            {"previous_status": "BLOCKED_EXTERNAL", "status": "ENTERPRISE_PASS"}
        ).encode(),
        json.dumps(
            {"historical_status": "PASS_LOCAL", "status": "ENTERPRISE_PASS"}
        ).encode(),
        json.dumps(
            {"prior_status": "BLOCKED_EXTERNAL", "status": "ENTERPRISE_PASS"}
        ).encode(),
        json.dumps(
            {"superseded_status": "PASS_LOCAL", "status": "ENTERPRISE_PASS"}
        ).encode(),
        json.dumps(
            {"history": {"status": "BLOCKED_EXTERNAL"}, "status": "ENTERPRISE_PASS"}
        ).encode(),
        b"| previous_status | notes |\n|---|---|\n| BLOCKED_EXTERNAL | superseded |",
        b"previous_status,notes\nPASS_LOCAL,superseded",
        b"\xff\xfe\x00\x89binary-evidence",
    ],
)
def test_structured_historical_or_binary_evidence_is_allowed(
    tmp_path: Path, content: bytes
) -> None:
    package = _accepted_bundle(tmp_path)
    _replace_first_case_evidence(package, content)
    assert validator.validate_acceptance_bundle(package)["status"] == "ACCEPTED"


@pytest.mark.parametrize(
    "content",
    [
        json.dumps({"current_status": "BLOCKED_EXTERNAL"}).encode(),
        json.dumps(
            {"status": "ENTERPRISE_PASS", "measurement": "TBD_EXTERNAL"}
        ).encode(),
        json.dumps({"priority": "TBD_EXTERNAL"}).encode(),
        json.dumps({"prioritization": "TBD_EXTERNAL"}).encode(),
        json.dumps({"原因": "TBD_EXTERNAL"}, ensure_ascii=False).encode(),
        json.dumps({"原值": "TBD_EXTERNAL"}, ensure_ascii=False).encode(),
        b"| status | notes |\n|---|---|\n| BLOCKED_EXTERNAL | still open |",
        b"| metric | measured_value |\n|---|---|\n| onboarding | TBD_EXTERNAL |",
        b"| priority | TBD_EXTERNAL |",
        b"state,notes\nPASS_LOCAL,still local",
        b"metric,measured_value\nonboarding,TBD_EXTERNAL",
        b"priority,TBD_EXTERNAL",
    ],
)
def test_structured_current_state_evidence_is_rejected(
    tmp_path: Path, content: bytes
) -> None:
    package = _accepted_bundle(tmp_path)
    _replace_first_case_evidence(package, content)
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("placeholder evidence" in failure for failure in result["failures"])


@pytest.mark.parametrize(
    ("path", "invalid_value"),
    [
        (("source_cases",), {"case": "P2-WP13-UAT-021"}),
        (("signoffs",), {"role": "product_owner"}),
        (("preproduction_rehearsal", "rto_target_minutes"), "30"),
        (("training_records", 0, "participants"), [None]),
        (("signoffs", 0, "person_name"), 123),
        (("metrics", 0, "numerator"), float("nan")),
        (("metrics", 0, "value"), float("inf")),
        (("preproduction_rehearsal", "rto_minutes"), float("nan")),
        (("preproduction_rehearsal", "rpo_minutes"), float("inf")),
    ],
)
def test_malformed_types_and_non_finite_numbers_fail_without_raising(
    tmp_path: Path, path: tuple[object, ...], invalid_value: object
) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    _set_nested(payload, path, invalid_value)
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert result["failures"]


def test_self_approval_and_five_party_proxy_signing_are_rejected(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload["uat_cases"][0]["approved_by_person_id"] = payload["uat_cases"][0][
        "executed_by_person_id"
    ]
    payload["signoffs"][1]["person_id"] = payload["signoffs"][0]["person_id"]
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("self-approval" in failure for failure in result["failures"])
    assert any("proxy signing" in failure for failure in result["failures"])


def test_open_p0_defect_blocks_acceptance(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload["defect_gate"]["p0_open"] = 1
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("p0_open" in failure for failure in result["failures"])


def test_open_p1_with_dual_written_decision_is_accepted(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload["defect_gate"]["p1_open"] = 2
    payload["defect_gate"]["open_p1_ids"] = ["P1-UAT-021-001", "P1-UAT-023-002"]
    payload["defect_gate"]["p1_disposition"].update(
        {
            "status": "WRITTEN_DECISION_APPROVED",
            "covered_p1_count": 2,
            "covered_p1_ids": ["P1-UAT-023-002", "P1-UAT-021-001"],
            "business_person_id": "p1-business-approver",
            "technical_person_id": "p1-technical-approver",
            "decided_at": "2026-08-04T14:30:00+08:00",
        }
    )
    package.write_text(json.dumps(payload), encoding="utf-8")
    assert validator.validate_acceptance_bundle(package)["status"] == "ACCEPTED"


def test_open_p1_without_written_decision_is_rejected(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload["defect_gate"]["p1_open"] = 1
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("WRITTEN_DECISION_APPROVED" in failure for failure in result["failures"])


def test_open_p1_ids_must_be_unique_and_exactly_covered(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload["defect_gate"]["p1_open"] = 2
    payload["defect_gate"]["open_p1_ids"] = ["P1-UAT-021-001", "P1-UAT-021-001"]
    payload["defect_gate"]["p1_disposition"].update(
        {
            "status": "WRITTEN_DECISION_APPROVED",
            "covered_p1_count": 2,
            "covered_p1_ids": ["P1-UAT-021-001", "TBD_EXTERNAL"],
            "business_person_id": "p1-business-approver",
            "technical_person_id": "p1-technical-approver",
            "decided_at": "2026-08-04T14:30:00+08:00",
        }
    )
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("unique P1 ids" in failure for failure in result["failures"])
    assert any("invalid or placeholder P1 id" in failure for failure in result["failures"])


@pytest.mark.parametrize(
    ("field_path", "wrong_value", "message"),
    [
        (("release_artifact", "version"), "v0.3.0-phase2-hotfix", "release_artifact.version"),
        (("execution_context", "executed_commit"), "5267d8bd8b30a58ed5fb6ec855ea579e7ce34d69", "executed_commit"),
    ],
)
def test_release_version_and_tag_commit_are_exact(
    tmp_path: Path, field_path: tuple[str, str], wrong_value: str, message: str
) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload[field_path[0]][field_path[1]] = wrong_value
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any(message in failure for failure in result["failures"])


def test_reversed_context_or_metric_window_is_rejected(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload["execution_context"]["started_at"] = "2026-08-05T19:00:00+08:00"
    payload["metrics"][0]["window_start"] = "2026-08-03T19:00:00+08:00"
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("started_at must not be after completed_at" in failure for failure in result["failures"])
    assert any("window_start must not be after window_end" in failure for failure in result["failures"])


def test_case_and_rehearsal_environment_must_match_context(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload["uat_cases"][0]["environment_id"] = "preprod-other"
    payload["preproduction_rehearsal"]["environment_id"] = "preprod-other"
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert sum("must match execution_context.environment_id" in failure for failure in result["failures"]) == 2


def test_signoff_cannot_precede_acceptance_completion_events(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload["signoffs"][0]["signed_at"] = "2026-08-03T10:00:00+08:00"
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("must not precede completion event" in failure for failure in result["failures"])


def test_incomplete_rehearsal_or_training_is_rejected(tmp_path: Path) -> None:
    package = _accepted_bundle(tmp_path)
    payload = json.loads(package.read_text())
    payload["preproduction_rehearsal"]["steps"][2]["status"] = "BLOCKED_EXTERNAL"
    payload["training_records"][0]["exercise_passed"] = False
    package.write_text(json.dumps(payload), encoding="utf-8")
    result = validator.validate_acceptance_bundle(package)
    assert result["status"] == "FAILED"
    assert any("rehearsal" in failure for failure in result["failures"])
    assert any("training" in failure for failure in result["failures"])