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",
    }
    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": "revoked"},
        {"rule_status": "deprecated"},
        {"plan_hash": "b" * 64},
    ],
)
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)