DataOps-platform/tests/acceptance/test_p2_wp12_second_domain_replication.py

from __future__ import annotations

import csv
import importlib.util
import json
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
PACKAGE_DIR = ROOT / "docs/phase2/p2-wp12-spare-parts-package"
MODULE_PATH = ROOT / "app/core/governance/domain_replication.py"
SPEC = importlib.util.spec_from_file_location("domain_replication", MODULE_PATH)
assert SPEC and SPEC.loader
domain_replication = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(domain_replication)


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


def _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 _accept(manifest=None, evidence=None):
    return domain_replication.evaluate_replication_package(
        manifest or _json("manifest.json"),
        evidence or _json("evidence.json"),
        template=_json("domain-template.json"),
        snapshot_rows=_rows("material_master_snapshot.csv"),
        delta_rows=_rows("material_master_delta.csv"),
    )


def test_second_domain_package_completes_all_platform_stages_without_raw_rows():
    report = _accept()

    assert report["status"] == "passed"
    assert report["domain"]["code"] == "spare_parts"
    assert report["enterprise_uat"]["status"] == "blocked_external"
    assert report["third_domain_reusable"] is True
    assert report["device_specific_core_changes"] == 0
    assert set(report["stages"]) == set(domain_replication.REQUIRED_STAGES)
    assert all(item["status"] == "passed" for item in report["stages"].values())
    serialized = json.dumps(report, ensure_ascii=False)
    assert "MAT-0001" not in serialized
    assert "轴承" not in serialized


def test_controlled_source_snapshot_and_delta_are_incremental_and_idempotent():
    report = _accept()["incremental_collection"]

    assert report["source_kind"] == "controlled_csv"
    assert report["classification"] == "desensitized"
    assert report["snapshot_rows"] == 3
    assert report["delta_rows"] == 2
    assert report["inserted"] == 1
    assert report["updated"] == 1
    assert report["deleted"] == 0
    assert report["final_rows"] == 4
    assert report["replay_changes"] == 0
    assert report["cursor_before"] == 100
    assert report["cursor_after"] == 102
    assert len(report["final_digest"]) == 64


def test_missing_or_failed_capability_evidence_blocks_acceptance():
    evidence = _json("evidence.json")
    evidence["receipts"] = [
        item for item in evidence["receipts"] if item["stage"] != "data_product"
    ]
    with pytest.raises(ValueError, match="missing stages.*data_product"):
        _accept(evidence=evidence)

    evidence = _json("evidence.json")
    quality = next(
        item for item in evidence["receipts"] if item["stage"] == "quality"
    )
    quality["status"] = "failed"
    with pytest.raises(ValueError, match="quality.*passed"):
        _accept(evidence=evidence)

    manifest = _json("manifest.json")
    manifest["enterprise_bindings"] = manifest["enterprise_bindings"][:-1]
    with pytest.raises(ValueError, match="missing enterprise bindings.*uat_users"):
        _accept(manifest=manifest)


def test_quality_remediation_observability_product_and_agent_are_closed_loop():
    report = _accept()
    quality = report["stages"]["quality"]["metrics"]
    remediation = report["stages"]["remediation"]["metrics"]
    observability = report["stages"]["observability"]["metrics"]
    product = report["stages"]["data_product"]["metrics"]
    agent = report["stages"]["agent"]["metrics"]

    assert quality["published_rule_count"] >= 5
    assert quality["final_score"] >= quality["target_score"]
    assert remediation == {
        "issue_status": "closed",
        "task_status": "completed",
        "independent_closer": True,
    }
    assert observability["incident_status"] == "closed"
    assert observability["recovered"] is True
    assert product["contract_status"] == "active"
    assert product["certificate_status"] == "issued"
    assert agent["autonomy_level"] in {"read_only", "suggestion"}
    assert agent["decision"] == "authorized"
    assert agent["automatic_execution_allowed"] is False
    assert agent["citation_count"] >= 1
    assert agent["cross_domain_denied"] is True


def test_device_specific_changes_and_device_api_evidence_are_rejected():
    manifest = _json("manifest.json")
    manifest["core_change_assessment"]["device_specific_changes"] = [
        "app/core/data_research/device_assets.py"
    ]
    with pytest.raises(ValueError, match="device-specific"):
        _accept(manifest=manifest)

    evidence = _json("evidence.json")
    evidence["receipts"][0]["api_refs"] = [
        "/api/development/v1/device-semantics/bootstrap"
    ]
    with pytest.raises(ValueError, match="device-specific API"):
        _accept(evidence=evidence)


def test_secret_like_source_columns_and_non_monotonic_cursors_are_rejected():
    snapshot = _rows("material_master_snapshot.csv")
    snapshot[0]["api_token"] = "not-allowed"
    with pytest.raises(ValueError, match="secret-like source field"):
        domain_replication.collect_incremental_rows(
            snapshot,
            _rows("material_master_delta.csv"),
            _json("manifest.json")["source"],
        )

    delta = _rows("material_master_delta.csv")
    delta[1]["_cursor"] = "100"
    with pytest.raises(ValueError, match="strictly increase"):
        domain_replication.collect_incremental_rows(
            _rows("material_master_snapshot.csv"),
            delta,
            _json("manifest.json")["source"],
        )


def test_product_certificate_and_agent_citations_must_bind_canonical_evidence():
    evidence = _json("evidence.json")
    product = next(
        item for item in evidence["receipts"] if item["stage"] == "data_product"
    )
    product["metrics"]["certificate_evidence_refs"] = []
    with pytest.raises(ValueError, match="certificate evidence"):
        _accept(evidence=evidence)

    evidence = _json("evidence.json")
    agent = next(item for item in evidence["receipts"] if item["stage"] == "agent")
    agent["metrics"]["citation_count"] = 0
    with pytest.raises(ValueError, match="citation"):
        _accept(evidence=evidence)