DataOps-platform/tests/mcp/test_canary_verifier.py

import pytest

from app.core.mcp.canary import CanaryVerifier


class Plans:
    def __init__(self):
        self.evidence = {
            "candidate_id": "candidate-1",
            "execution_id": "execution-1",
            "status": "started",
            "sample_runs": 0,
        }
        self.recorded = []

    def get_canary_evidence(self, candidate_id):
        assert candidate_id == "candidate-1"
        return dict(self.evidence)

    def record_canary_verification(self, candidate_id, evidence):
        self.recorded.append((candidate_id, evidence))
        self.evidence = dict(evidence)
        return evidence


class Engine:
    def __init__(self, state):
        self.state = state

    def get_execution(self, execution_id):
        assert execution_id == "execution-1"
        return {
            "id": execution_id,
            "state": {"current": self.state},
            "outputs": {"secret": "must-not-persist"},
        }


class Comparator:
    def __init__(self, equivalent=True):
        self.equivalent = equivalent

    def compare(self, baseline_execution_id, candidate_execution_id):
        assert baseline_execution_id == "baseline-1"
        assert candidate_execution_id == "execution-1"
        return {
            "equivalent": self.equivalent,
            "metrics": {
                "row_count_delta": 0,
                "duration_delta_seconds": -2,
            },
            "raw_rows": [{"email": "must-not-persist"}],
        }


def test_canary_verifier_records_pass_only_from_successful_engine_and_comparator():
    plans = Plans()
    verifier = CanaryVerifier(
        plans=plans,
        engine=Engine("SUCCESS"),
        comparator=Comparator(equivalent=True),
    )

    result = verifier.verify("candidate-1", baseline_execution_id="baseline-1")

    assert result["status"] == "passed"
    assert result["sample_runs"] == 1
    assert result["verified_by"] == "dataops-canary-verifier"
    assert result["verification_summary"] == {
        "engine_state": "SUCCESS",
        "equivalent": True,
        "metrics": {
            "row_count_delta": 0,
            "duration_delta_seconds": -2,
        },
    }
    assert "must-not-persist" not in repr(plans.recorded)


def test_canary_verifier_fails_closed_for_non_terminal_or_non_equivalent_runs():
    plans = Plans()
    running = CanaryVerifier(plans=plans, engine=Engine("RUNNING"))

    result = running.verify("candidate-1")
    assert result["status"] == "started"
    assert plans.recorded == []

    failed = CanaryVerifier(
        plans=Plans(),
        engine=Engine("SUCCESS"),
        comparator=Comparator(equivalent=False),
    )
    result = failed.verify("candidate-1", baseline_execution_id="baseline-1")
    assert result["status"] == "failed"


def test_canary_verifier_rejects_untrusted_preverified_evidence():
    plans = Plans()
    plans.evidence["status"] = "passed"
    verifier = CanaryVerifier(plans=plans, engine=Engine("SUCCESS"))

    with pytest.raises(ValueError, match="already finalized"):
        verifier.verify("candidate-1")