DataOps-platform/tests/integration/test_data_research_reconciliation.py

from __future__ import annotations


class ControlPlane:
    def published_versions(self):
        return [
            {"ontology_uid": "ontology-1", "version_uid": "version-2", "content_hash": "hash-2"}
        ]

    def artifacts(self):
        return [
            {"uid": "artifact-1", "storage_ref": "minio://bucket/present"},
            {"uid": "artifact-2", "storage_ref": "minio://bucket/missing"},
        ]

    def governance_documents(self):
        return [{"object_uid": "ontology-1", "object_version": 2, "content_hash": "old-hash"}]


class Graph:
    def active_versions(self):
        return {"ontology-1": "version-1"}

    def repair_projection(self, issue):
        self.repaired = issue


class Storage:
    def exists(self, ref):
        return not ref.endswith("missing")


class Knowledge:
    def expected_hash(self, ontology_uid, version_uid):
        return "hash-2"

    def repair_document(self, issue):
        self.repaired = issue


def test_read_only_reconciliation_builds_explicit_repair_plan():
    from app.core.data_research.operations import DataResearchReconciler

    graph = Graph()
    knowledge = Knowledge()
    reconciler = DataResearchReconciler(ControlPlane(), graph, Storage(), knowledge)
    report = reconciler.reconcile(repair=False)

    assert {issue["type"] for issue in report["issues"]} == {
        "projection_mismatch",
        "artifact_missing",
        "knowledge_mismatch",
    }
    assert report["repair_mode"] is False
    assert not hasattr(graph, "repaired") and not hasattr(knowledge, "repaired")


def test_explicit_repair_mode_repairs_rebuildable_projections_only():
    from app.core.data_research.operations import DataResearchReconciler

    graph = Graph()
    knowledge = Knowledge()
    report = DataResearchReconciler(ControlPlane(), graph, Storage(), knowledge).reconcile(
        repair=True
    )
    assert report["repaired"] == 2
    assert graph.repaired["type"] == "projection_mismatch"
    assert knowledge.repaired["type"] == "knowledge_mismatch"
    assert all(issue["type"] != "artifact_missing" for issue in report["repair_plan"])