DataOps-platform/tests/runner/test_rules.py

from __future__ import annotations

import copy
import hashlib
import json

import pytest

from app.core.common.identifiers import new_governance_uid
from app.runner.nodes import NodeExecutionError

PLAN = {"op": "not_null", "column": "mobile"}
PLAN_HASH = hashlib.sha256(
    json.dumps(
        PLAN,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    ).encode("utf-8")
).hexdigest()


def rule_node(node_type="quality.check"):
    node = {
        "id": "customer_mobile",
        "type": node_type,
        "purpose": "read" if node_type == "quality.check" else "write",
        "config": {
            "component_binding_id": new_governance_uid(),
            "rule_version_id": new_governance_uid(),
            "execution_plan_hash": PLAN_HASH,
        },
    }
    if node_type == "rule.apply":
        node["idempotency"] = {
            "strategy": "upsert",
            "key": "customer_id",
        }
    return node


class Repository:
    def __init__(self, record=None):
        self.record = record
        self.calls = []

    def load(self, **kwargs):
        self.calls.append(kwargs)
        return copy.deepcopy(self.record)


class Adapter:
    def __init__(self):
        self.calls = []

    def execute(self, *, plan, node, parameters, write_authorized):
        self.calls.append(
            {
                "plan": plan,
                "node": node,
                "parameters": parameters,
                "write_authorized": write_authorized,
            }
        )
        return {"rows_rejected": 3}


def published_record(node, **overrides):
    value = {
        "component_binding_id": node["config"]["component_binding_id"],
        "rule_version_id": node["config"]["rule_version_id"],
        "backend": "quality_check",
        "plan": PLAN,
        "plan_hash": PLAN_HASH,
        "plan_status": "published",
        "rule_status": "published",
        "publication_audit_trusted": True,
        "logical_evidence_trusted": True,
        "physical_evidence_trusted": True,
        "component_kind": node["type"],
        "binding_idempotency": node.get("idempotency"),
    }
    value.update(overrides)
    return value


def test_rule_executor_loads_only_published_plan_by_fixed_identifiers():
    from app.runner.rules import RulePlanExecutor

    node = rule_node()
    adapter = Adapter()
    repository = Repository(published_record(node))
    executor = RulePlanExecutor(
        repository,
        adapters={"quality_check": adapter},
    )

    result = executor.execute(node, {"partition": "2026-07-23"})

    assert result["rows_rejected"] == 3
    assert result["rule_version_id"] == node["config"]["rule_version_id"]
    assert repository.calls == [
        {
            "component_binding_id": node["config"]["component_binding_id"],
            "rule_version_id": node["config"]["rule_version_id"],
            "plan_hash": PLAN_HASH,
        }
    ]
    assert adapter.calls[0]["parameters"] == {"partition": "2026-07-23"}


@pytest.mark.parametrize(
    "record",
    [
        None,
        {"plan_status": "compiled"},
        {"plan_status": "tested"},
        {"plan_status": "revoked"},
        {"rule_status": "deprecated"},
        {"plan_hash": "b" * 64},
        {"publication_audit_trusted": False},
        {"logical_evidence_trusted": False},
        {"physical_evidence_trusted": False},
    ],
)
def test_rule_executor_fails_closed_for_missing_revoked_or_mismatched_plan(record):
    from app.runner.rules import RulePlanExecutor

    node = rule_node()
    base = published_record(node)
    if record is not None:
        base.update(record)
        record = base
    executor = RulePlanExecutor(
        Repository(record),
        adapters={"quality_check": Adapter()},
    )

    with pytest.raises(NodeExecutionError):
        executor.execute(node, {})


def test_rule_executor_rejects_inline_plan_or_unregistered_backend():
    from app.runner.rules import RulePlanExecutor

    node = rule_node()
    node["config"]["plan"] = {"op": "bypass"}
    executor = RulePlanExecutor(
        Repository(published_record(node)),
        adapters={},
    )

    with pytest.raises(NodeExecutionError):
        executor.execute(node, {})

    clean = rule_node()
    with pytest.raises(NodeExecutionError):
        RulePlanExecutor(
            Repository(published_record(clean, backend="generated_python")),
            adapters={},
        ).execute(clean, {})


def test_mutating_rule_requires_governed_write_authorization_and_idempotency():
    from app.runner.rules import RulePlanExecutor

    node = rule_node("rule.apply")
    record = published_record(node, backend="sql_pushdown")
    executor = RulePlanExecutor(
        Repository(record),
        adapters={"sql_pushdown": Adapter()},
    )

    with pytest.raises(NodeExecutionError):
        executor.execute(node, {}, write_authorized=False)

    del node["idempotency"]
    with pytest.raises(NodeExecutionError):
        executor.execute(node, {}, write_authorized=True)


def test_quality_node_dispatches_sql_plan_to_read_only_quality_adapter():
    from app.core.data_rules.compilers.sql import COMPILER_VERSION
    from app.runner.rules import RulePlanExecutor
    from tests.runner.test_rule_sql import sql_plan

    plan, plan_hash = sql_plan()
    node = rule_node()
    node["config"]["rule_version_id"] = plan["rule_version_id"]
    node["config"]["execution_plan_hash"] = plan_hash
    quality_adapter = Adapter()
    record = published_record(
        node,
        backend="sql_pushdown",
        compiler_version=COMPILER_VERSION,
        plan=plan,
        plan_hash=plan_hash,
        schema_hashes={
            "rule_spec_hash": plan["rule_spec_hash"],
            "input_schema_snapshot_id": plan[
                "input_schema_snapshot_id"
            ],
            "input_schema_hash": plan["input_schema_hash"],
            "output_schema_snapshot_id": plan[
                "output_schema_snapshot_id"
            ],
            "output_schema_hash": plan["output_schema_hash"],
        },
        canonical_rule_spec_hash=plan["rule_spec_hash"],
        canonical_input_schema_snapshot_id=plan[
            "input_schema_snapshot_id"
        ],
        canonical_input_schema_hash=plan["input_schema_hash"],
        canonical_output_schema_snapshot_id=plan[
            "output_schema_snapshot_id"
        ],
        canonical_output_schema_hash=plan["output_schema_hash"],
        canonical_input_data_source_uid=plan["data_source_uid"],
        canonical_output_data_source_uid=plan["data_source_uid"],
        canonical_input_object_ref="raw.customer",
        canonical_output_object_ref="clean.customer",
        canonical_input_dialect="postgresql",
        canonical_output_dialect="postgresql",
        canonical_input_access_mode="read",
        canonical_input_write_mode=None,
        canonical_output_access_mode="write",
        canonical_output_write_mode="append",
    )
    repository = Repository(record)
    class Evidence:
        heartbeat_interval_seconds = 10

        def __init__(self):
            self.resolved = []

        def start(self, **_kwargs):
            return new_governance_uid()

        def replay(self, _rule_run_id):
            return None

        def heartbeat(self, _rule_run_id, _lease_owner):
            return None

        def finish(self, _rule_run_id, _result):
            return None

        def resolve_sql_staging(self, receipt, **kwargs):
            self.resolved.append((receipt, kwargs))

    evidence = Evidence()
    executor = RulePlanExecutor(
        repository,
        adapters={"quality_check": quality_adapter},
        evidence_writer=evidence,
    )

    context = {
        "correlation_id": new_governance_uid(),
        "dataflow_uid": new_governance_uid(),
        "deployment_id": new_governance_uid(),
        "environment": "test",
        "workflow_version": 1,
        "node_id": node["id"],
        "task_jti": new_governance_uid(),
    }
    receipt = (
        "dataops-staging://01900000-0000-7000-8000-000000000099"
    )
    result = executor.execute(
        node,
        {"input_artifact": receipt},
        **context,
    )

    assert result["execution_plan_hash"] == plan_hash
    assert quality_adapter.calls[0]["write_authorized"] is False
    assert quality_adapter.calls[0]["parameters"] == {}
    assert evidence.resolved == [
        (
            receipt,
            {
                "deployment_id": context["deployment_id"],
                "correlation_id": context["correlation_id"],
                "input_binding_id": plan["input_binding_id"],
            },
        )
    ]


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("canonical_input_access_mode", "write"),
        ("canonical_input_write_mode", "append"),
        ("canonical_output_access_mode", "read"),
        ("canonical_output_write_mode", None),
    ],
)
def test_sql_plan_rechecks_live_binding_access_before_execution(field, value):
    from app.core.data_rules.compilers.sql import COMPILER_VERSION
    from app.runner.rules import RulePlanExecutor
    from tests.runner.test_rule_sql import sql_plan

    plan, plan_hash = sql_plan()
    node = rule_node("rule.apply")
    node["config"]["rule_version_id"] = plan["rule_version_id"]
    node["config"]["execution_plan_hash"] = plan_hash
    values = {
        "backend": "sql_pushdown",
        "compiler_version": COMPILER_VERSION,
        "plan": plan,
        "plan_hash": plan_hash,
        "schema_hashes": {
            "rule_spec_hash": plan["rule_spec_hash"],
            "input_schema_snapshot_id": plan["input_schema_snapshot_id"],
            "input_schema_hash": plan["input_schema_hash"],
            "output_schema_snapshot_id": plan["output_schema_snapshot_id"],
            "output_schema_hash": plan["output_schema_hash"],
        },
        "canonical_rule_spec_hash": plan["rule_spec_hash"],
        "canonical_input_schema_snapshot_id": plan["input_schema_snapshot_id"],
        "canonical_input_schema_hash": plan["input_schema_hash"],
        "canonical_output_schema_snapshot_id": plan["output_schema_snapshot_id"],
        "canonical_output_schema_hash": plan["output_schema_hash"],
        "canonical_input_data_source_uid": plan["data_source_uid"],
        "canonical_output_data_source_uid": plan["data_source_uid"],
        "canonical_input_object_ref": "raw.customer",
        "canonical_output_object_ref": "clean.customer",
        "canonical_input_dialect": "postgresql",
        "canonical_output_dialect": "postgresql",
        "canonical_input_access_mode": "read",
        "canonical_input_write_mode": None,
        "canonical_output_access_mode": "write",
        "canonical_output_write_mode": "append",
    }
    values[field] = value
    record = published_record(node, **values)

    with pytest.raises(
        NodeExecutionError,
        match="canonical attestation",
    ):
        RulePlanExecutor(
            Repository(record),
            adapters={"sql_pushdown": Adapter()},
        ).execute(node, {}, write_authorized=True)