DataOps-platform/tests/acceptance/test_phase3_wp08_third_domain_replication.py

from __future__ import annotations

import csv
import hashlib
import importlib.util
import json
import os
import shutil
from copy import deepcopy
from pathlib import Path

import pytest

import app.core.governance.domain_replication_contract_registry as contract_registry
from app.core.governance.domain_replication import (
    REQUIRED_STAGES,
    DomainReplicationError,
    validate_contract_reference,
    validate_execution_evidence,
)
from app.core.governance.domain_replication_contract_registry import registry_snapshot

ROOT = Path(__file__).resolve().parents[2]
PACKAGE_DIR = ROOT / "docs/phase3/p3-wp08-third-domain-template"
P2_PACKAGE_DIR = ROOT / "docs/phase2/p2-wp12-spare-parts-package"
VALIDATOR_PATH = ROOT / "scripts/validate_domain_replication.py"
VALIDATOR_SPEC = importlib.util.spec_from_file_location("wp08_validator", VALIDATOR_PATH)
assert VALIDATOR_SPEC and VALIDATOR_SPEC.loader
validator = importlib.util.module_from_spec(VALIDATOR_SPEC)
VALIDATOR_SPEC.loader.exec_module(validator)


def _package_json(name: str) -> dict:
    return json.loads((PACKAGE_DIR / name).read_text(encoding="utf-8"))


def _package_rows(name: str) -> list[dict[str, str]]:
    with (PACKAGE_DIR / "data" / name).open(encoding="utf-8", newline="") as handle:
        return list(csv.DictReader(handle))


def _package_copy(tmp_path: Path) -> Path:
    package = tmp_path / "package"
    shutil.copytree(PACKAGE_DIR, package)
    return package


def _execution_evidence() -> dict:
    return {
        "schema_version": 1,
        "stages": [
            {
                "stage": stage,
                "actions": [
                    {
                        "action": "dry_run",
                        "status": "passed",
                        "write_count": 0,
                        "evidence_ref": f"dry-run://third_domain_template/{stage}",
                        "audit_ref": f"audit://third_domain_template/{stage}/dry-run",
                    },
                    {
                        "action": "execute",
                        "status": "passed",
                        "write_count": 1,
                        "evidence_ref": f"execution://third_domain_template/{stage}",
                        "audit_ref": f"audit://third_domain_template/{stage}/execute",
                    },
                    {
                        "action": "diff",
                        "status": "passed",
                        "write_count": 0,
                        "evidence_ref": f"diff://third_domain_template/{stage}",
                        "audit_ref": f"audit://third_domain_template/{stage}/diff",
                    },
                ],
            }
            for stage in REQUIRED_STAGES
        ],
    }


def _strict_execution_evidence() -> dict:
    """V2 execution receipts bind rollback to its executed state transition."""

    evidence = _execution_evidence()
    evidence["schema_version"] = 2
    replay_receipts: list[dict[str, object]] = []
    for index, receipt in enumerate(evidence["stages"]):
        stage = receipt["stage"]
        execute = receipt["actions"][1]
        execute.update(
            {
                "operation_uid": f"execute-{index:02d}",
                "idempotency_key": f"execute-{index:02d}",
                "request_digest": f"{index + 1:064x}",
                "attempt": 1,
                "lease_fence": index + 1,
                "pre_state_digest": f"{index + 101:064x}",
                "post_state_digest": f"{index + 201:064x}",
                "result_digest": f"{index + 301:064x}",
            }
        )
        if stage == "remediation":
            rollback = receipt["actions"][2]
            rollback.update(
                {
                    "action": "rollback",
                    "operation": "restore",
                    "operation_uid": execute["operation_uid"],
                    "idempotency_key": execute["idempotency_key"],
                    "request_digest": execute["request_digest"],
                    "attempt": execute["attempt"],
                    "lease_fence": execute["lease_fence"],
                    "from_digest": execute["post_state_digest"],
                    "to_digest": execute["pre_state_digest"],
                    "result_digest": f"{index + 401:064x}",
                    "status": "passed",
                }
            )
            replay_receipts.append(
                {
                    "stage": stage,
                    "operation_uid": execute["operation_uid"],
                    "idempotency_key": execute["idempotency_key"],
                    "request_digest": execute["request_digest"],
                    "attempt": execute["attempt"],
                    "lease_fence": execute["lease_fence"],
                    "from_digest": execute["post_state_digest"],
                    "to_digest": execute["pre_state_digest"],
                    "result_digest": rollback["result_digest"],
                    "status": "exact_replay",
                }
            )
    evidence["replay_receipts"] = replay_receipts
    return evidence


def test_v2_rollback_requires_matching_execute_state_and_independent_replay_receipt():
    execution = _strict_execution_evidence()

    assert validate_execution_evidence(execution)["remediation"]["recovery_strategy"] == "rollback"

    delayed = deepcopy(execution)
    delayed["stages"][6]["actions"][2]["lease_fence"] = 6
    with pytest.raises(DomainReplicationError, match="lease_fence"):
        validate_execution_evidence(delayed)

    changed = deepcopy(execution)
    changed["replay_receipts"][0]["to_digest"] = "a" * 64
    with pytest.raises(DomainReplicationError, match="replay receipt"):
        validate_execution_evidence(changed)

    forged = deepcopy(execution)
    forged["stages"][6]["actions"][2]["from_digest"] = "b" * 64
    with pytest.raises(DomainReplicationError, match="execute post_state_digest"):
        validate_execution_evidence(forged)

    malformed = deepcopy(execution)
    malformed["stages"][6]["actions"][2]["result_digest"] = "not-a-digest"
    malformed["replay_receipts"][0]["result_digest"] = "not-a-digest"
    with pytest.raises(DomainReplicationError, match="remediation.rollback.result_digest"):
        validate_execution_evidence(malformed)


def test_execution_evidence_requires_ordered_dry_run_execute_and_diff():
    summary = validate_execution_evidence(_execution_evidence())

    assert list(summary) == list(REQUIRED_STAGES)
    assert all(item["actions"] == ["dry_run", "execute", "diff"] for item in summary.values())
    assert all(item["dry_run_write_count"] == 0 for item in summary.values())
    assert all(item["recovery_strategy"] == "diff" for item in summary.values())
    assert all(len(item["evidence_digest"]) == 64 for item in summary.values())


@pytest.mark.parametrize(
    ("mutate", "message"),
    [
        (
            lambda evidence: evidence["stages"][0]["actions"].__setitem__(
                0, evidence["stages"][0]["actions"][1]
            ),
            "ordered",
        ),
        (
            lambda evidence: evidence["stages"][0]["actions"][0].update(
                {"write_count": 1}
            ),
            "dry_run",
        ),
        (
            lambda evidence: evidence["stages"][0]["actions"][2].update(
                {"action": "verify"}
            ),
            "diff or rollback",
        ),
        (
            lambda evidence: evidence["stages"][0]["actions"][1].update(
                {"source_url": "https://enterprise.invalid/source"}
            ),
            "URL",
        ),
        (
            lambda evidence: evidence["stages"][0]["actions"][1].update(
                {"raw_rows": [{"id": "sample"}]}
            ),
            "raw source rows",
        ),
    ],
)
def test_execution_evidence_fails_closed_for_unsafe_or_incomplete_receipts(
    mutate, message: str
):
    evidence = _execution_evidence()
    mutate(evidence)

    with pytest.raises(DomainReplicationError, match=message):
        validate_execution_evidence(evidence)


def test_execution_evidence_rejects_cross_domain_receipt_reference():
    evidence = _execution_evidence()
    evidence["stages"][1]["actions"][1]["evidence_ref"] = (
        "execution://other_domain/catalog"
    )

    with pytest.raises(DomainReplicationError, match="cross-domain"):
        validate_execution_evidence(evidence)


def test_execution_evidence_is_deterministic_across_a_replay():
    evidence = _execution_evidence()

    assert validate_execution_evidence(evidence) == validate_execution_evidence(
        deepcopy(evidence)
    )


def test_third_domain_template_is_hash_bound_and_uat_blocked():
    manifest = _package_json("manifest.json")
    evidence = _package_json("evidence.json")
    execution_evidence = _package_json("execution-evidence.json")

    report = __import__(
        "app.core.governance.domain_replication", fromlist=["evaluate_replication_package"]
    ).evaluate_replication_package(
        manifest,
        evidence,
        template=_package_json("domain-template.json"),
        snapshot_rows=_package_rows("governance_object_snapshot.csv"),
        delta_rows=_package_rows("governance_object_delta.csv"),
        execution_evidence=execution_evidence,
    )

    assert report["schema_version"] == 2
    assert report["status"] == "passed"
    assert report["enterprise_uat"] == {
        "status": "blocked_external",
        "unbound_requirements": [
            "data_steward",
            "domain_owner",
            "readonly_source",
            "uat_users",
        ],
    }
    assert set(report["execution_evidence"]) == set(REQUIRED_STAGES)
    assert report["device_specific_core_changes"] == 0
    assert "spare_parts" not in json.dumps(report, ensure_ascii=False)
    assert "/device-" not in json.dumps(report, ensure_ascii=False)
    assert "sample-001" not in json.dumps(report, ensure_ascii=False)

    for binding in manifest["files"].values():
        path = PACKAGE_DIR / binding["path"]
        assert path.is_file()
        assert hashlib.sha256(path.read_bytes()).hexdigest() == binding["sha256"]
    assert json.loads((PACKAGE_DIR / "acceptance-report.json").read_text(encoding="utf-8")) == report


@pytest.mark.parametrize("path_name", ["governance_object_snapshot.csv", "manifest.json"])
def test_package_reader_rejects_hardlinked_inputs(tmp_path: Path, path_name: str):
    package = _package_copy(tmp_path)
    target = next(path for path in package.rglob(path_name))
    os.link(target, package / f"second-{path_name}")

    with pytest.raises(DomainReplicationError, match="hardlink"):
        validator.validate_package(package)


def test_package_reader_rejects_symlinked_input(tmp_path: Path):
    package = _package_copy(tmp_path)
    snapshot = package / "data/governance_object_snapshot.csv"
    replacement = package / "snapshot-safe-copy.csv"
    snapshot.rename(replacement)
    snapshot.symlink_to(replacement.name)

    with pytest.raises(DomainReplicationError, match="symlink"):
        validator.validate_package(package)


def test_package_reader_fails_closed_when_open_target_is_replaced(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
):
    package = _package_copy(tmp_path)
    original_open = os.open
    swapped = False

    def race_open(path, flags, *args, **kwargs):
        nonlocal swapped
        if path == "governance_object_snapshot.csv" and not swapped:
            swapped = True
            snapshot = package / "data/governance_object_snapshot.csv"
            replacement = package / "replacement.csv"
            snapshot.rename(replacement)
            snapshot.symlink_to(replacement.name)
        return original_open(path, flags, *args, **kwargs)

    monkeypatch.setattr(validator.os, "open", race_open)
    with pytest.raises(DomainReplicationError, match="symlink|changed"):
        validator.validate_package(package)


@pytest.mark.parametrize(
    ("target", "key", "value", "message"),
    [
        ("manifest", "unknown_field", True, "unsupported fields"),
        ("manifest", "tоken", "x", "sensitive"),
        ("manifest", "rаw_rows", [{"id": "sample"}], "raw source"),
        ("template", "records", [{"id": "sample"}], "unsupported fields"),
        ("evidence", "raw_rows", [{"id": "sample"}], "raw source"),
        ("evidence", "source_url", "https://enterprise.invalid", "URL"),
    ],
)
def test_closed_package_schemas_reject_unknown_confusable_and_unsafe_values(
    target: str, key: str, value: object, message: str
):
    manifest = _package_json("manifest.json")
    evidence = _package_json("evidence.json")
    template = _package_json("domain-template.json")
    {"manifest": manifest, "evidence": evidence, "template": template}[target][key] = value

    with pytest.raises(DomainReplicationError, match=message):
        __import__(
            "app.core.governance.domain_replication", fromlist=["evaluate_replication_package"]
        ).evaluate_replication_package(
            manifest,
            evidence,
            template=template,
            snapshot_rows=_package_rows("governance_object_snapshot.csv"),
            delta_rows=_package_rows("governance_object_delta.csv"),
            execution_evidence=_package_json("execution-evidence.json"),
        )


def test_terminal_evidence_requires_allowlisted_api_and_typed_connector_contract():
    evidence = _package_json("evidence.json")
    incremental = next(item for item in evidence["receipts"] if item["stage"] == "incremental_ingestion")
    incremental["api_refs"] = ["not-a-contract"]

    with pytest.raises(DomainReplicationError, match="allowlisted API"):
        _accept_third_domain(evidence=evidence)

    evidence = _package_json("evidence.json")
    incremental = next(item for item in evidence["receipts"] if item["stage"] == "incremental_ingestion")
    incremental["contract_refs"][0]["run_id"] = "not-a-contract"
    with pytest.raises(DomainReplicationError, match="trusted registry"):
        _accept_third_domain(evidence=evidence)


def test_closed_schema_rejects_terminal_unknown_field_and_depth_or_size_attack():
    evidence = _package_json("evidence.json")
    evidence["receipts"][0]["unexpected"] = True
    with pytest.raises(DomainReplicationError, match="unsupported fields"):
        _accept_third_domain(evidence=evidence)

    manifest = _package_json("manifest.json")
    nested: dict[str, object] = {"leaf": "safe"}
    for _ in range(13):
        nested = {"safe": nested}
    manifest["safe"] = nested
    with pytest.raises(DomainReplicationError, match="maximum package depth"):
        __import__(
            "app.core.governance.domain_replication", fromlist=["evaluate_replication_package"]
        ).evaluate_replication_package(
            manifest,
            _package_json("evidence.json"),
            template=_package_json("domain-template.json"),
            snapshot_rows=_package_rows("governance_object_snapshot.csv"),
            delta_rows=_package_rows("governance_object_delta.csv"),
            execution_evidence=_package_json("execution-evidence.json"),
        )

    evidence = _package_json("evidence.json")
    evidence["safe"] = "x" * 4097
    with pytest.raises(DomainReplicationError, match="maximum string length"):
        _accept_third_domain(evidence=evidence)


def test_rollback_receipt_binds_replay_fence_and_digests():
    execution = _execution_evidence()
    rollback = execution["stages"][0]["actions"][2]
    rollback.update(
        {
            "action": "rollback",
            "operation": "restore",
            "idempotency_key": "rollback-001",
            "request_digest": "a" * 64,
            "attempt": 1,
            "lease_fence": 4,
            "from_digest": "b" * 64,
            "to_digest": "c" * 64,
            "replay_status": "exact_replay",
        }
    )

    assert validate_execution_evidence(execution)["template_initialization"]["recovery_strategy"] == "rollback"

    rollback["lease_fence"] = 0
    with pytest.raises(DomainReplicationError, match="lease_fence"):
        validate_execution_evidence(execution)

    rollback["lease_fence"] = 4
    rollback["replay_status"] = "changed_replay_conflict"
    with pytest.raises(DomainReplicationError, match="exact_replay"):
        validate_execution_evidence(execution)


def test_v2_contract_reference_must_match_trusted_registry_run():
    registry = registry_snapshot()
    contract = registry["contracts"][0]
    run = contract["runs"][0]
    reference = {
        "contract_id": contract["contract_id"],
        "contract_version": contract["contract_version"],
        "schema_digest": contract["schema_digest"],
        **run,
    }

    assert validate_contract_reference(reference, expected_contract_id=contract["contract_id"]) == reference

    for key, value in (("run_id", "invented-run"), ("operation", "unknown"), ("schema_digest", "f" * 64)):
        forged = deepcopy(reference)
        forged[key] = value
        with pytest.raises(DomainReplicationError, match="registry|contract"):
            validate_contract_reference(forged, expected_contract_id=contract["contract_id"])


def test_trusted_registry_rejects_tamper_expiry_and_contract_version_mismatch(monkeypatch):
    original_registry = registry_snapshot()
    original_digest = contract_registry.REGISTRY_SHA256
    contract = registry_snapshot()["contracts"][0]
    reference = {
        "contract_id": contract["contract_id"],
        "contract_version": contract["contract_version"],
        "schema_digest": contract["schema_digest"],
        **contract["runs"][0],
    }
    tampered = registry_snapshot()
    tampered["contracts"][0]["allowed_operations"] = ["catalog"]
    monkeypatch.setattr(contract_registry, "CONTRACT_REGISTRY", tampered)
    with pytest.raises(DomainReplicationError, match="integrity"):
        validate_contract_reference(reference, expected_contract_id=contract["contract_id"])

    expired = registry_snapshot()
    expired["expires_on"] = "2000-01-01"
    monkeypatch.setattr(contract_registry, "CONTRACT_REGISTRY", expired)
    monkeypatch.setattr(
        contract_registry,
        "REGISTRY_SHA256",
        hashlib.sha256(contract_registry._canonical(expired)).hexdigest(),
    )
    with pytest.raises(DomainReplicationError, match="expired"):
        validate_contract_reference(reference, expected_contract_id=contract["contract_id"])

    monkeypatch.setattr(contract_registry, "CONTRACT_REGISTRY", original_registry)
    monkeypatch.setattr(contract_registry, "REGISTRY_SHA256", original_digest)
    mismatch = deepcopy(reference)
    mismatch["contract_version"] = "999"
    with pytest.raises(DomainReplicationError, match="version"):
        validate_contract_reference(mismatch, expected_contract_id=contract["contract_id"])


def test_v1_p2_golden_package_is_accepted_without_v2_contract_refs(tmp_path: Path):
    package = tmp_path / "p2-v1-copy"
    shutil.copytree(P2_PACKAGE_DIR, package)

    report = validator.validate_package(package)

    assert report["schema_version"] == 1
    assert all("contract_refs" not in stage for stage in report["stages"].values())


def test_v2_seed_items_reject_unknown_nested_object_field():
    template = _package_json("domain-template.json")
    template["seed_data"][0]["items"][0]["unexpected"] = True

    with pytest.raises(DomainReplicationError, match="unsupported fields"):
        _accept_third_domain(evidence=_package_json("evidence.json"), template=template)


def _accept_third_domain(*, evidence: dict, template: dict | None = None):
    return __import__(
        "app.core.governance.domain_replication", fromlist=["evaluate_replication_package"]
    ).evaluate_replication_package(
        _package_json("manifest.json"),
        evidence,
        template=template or _package_json("domain-template.json"),
        snapshot_rows=_package_rows("governance_object_snapshot.csv"),
        delta_rows=_package_rows("governance_object_delta.csv"),
        execution_evidence=_package_json("execution-evidence.json"),
    )