248 lines
8.5 KiB
Python
248 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.concurrency import RevisionConflictError
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_dataflow.backend.db.models import (
|
|
DataflowPipeline,
|
|
DataflowReconciliationDecision,
|
|
DataflowReconciliationDecisionSet,
|
|
)
|
|
from govoplan_dataflow.backend.executor import PipelineExecutionError
|
|
from govoplan_dataflow.backend.reconciliation_decisions import (
|
|
create_decision_set,
|
|
current_decision_rows,
|
|
current_decisions,
|
|
decision_set_fingerprint,
|
|
decision_set_payload,
|
|
decision_set_source_payload,
|
|
record_decision,
|
|
)
|
|
from govoplan_dataflow.backend.schemas import GraphNode, GraphPosition
|
|
from govoplan_dataflow.backend.service import _datasource_source_resolver
|
|
|
|
|
|
TABLES = (
|
|
DataflowPipeline.__table__,
|
|
DataflowReconciliationDecisionSet.__table__,
|
|
DataflowReconciliationDecision.__table__,
|
|
)
|
|
|
|
|
|
class ReconciliationDecisionTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(self.engine, tables=TABLES)
|
|
self.session = Session(self.engine)
|
|
self.pipeline = DataflowPipeline(
|
|
tenant_id="tenant-1",
|
|
scope_type="tenant",
|
|
scope_id="tenant-1",
|
|
name="Monthly reconciliation",
|
|
status="active",
|
|
created_by="account-1",
|
|
updated_by="account-1",
|
|
)
|
|
self.session.add(self.pipeline)
|
|
self.session.flush()
|
|
self.principal = ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset(
|
|
{
|
|
"dataflow:pipeline:read",
|
|
"dataflow:pipeline:write",
|
|
"dataflow:pipeline:run",
|
|
}
|
|
),
|
|
),
|
|
account=SimpleNamespace(id="account-1"),
|
|
user=SimpleNamespace(id="membership-1"),
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_decisions_are_immutable_current_rows_and_occ_protected(self) -> None:
|
|
decision_set = create_decision_set(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
pipeline_id=self.pipeline.id,
|
|
name="July review",
|
|
node_id="reconcile",
|
|
actor_ref="account:account-1",
|
|
)
|
|
first = record_decision(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
decision_set_id=decision_set.id,
|
|
expected_revision=1,
|
|
key_hash="a" * 64,
|
|
input_hash="b" * 64,
|
|
action="accept",
|
|
reason="Checked against the source record.",
|
|
correction=None,
|
|
actor_ref="account:account-1",
|
|
)
|
|
first_fingerprint = decision_set_fingerprint(first)
|
|
second = record_decision(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
decision_set_id=decision_set.id,
|
|
expected_revision=2,
|
|
key_hash="a" * 64,
|
|
input_hash="c" * 64,
|
|
action="correct",
|
|
reason="The monthly source changed after review.",
|
|
correction={"amount": 25},
|
|
actor_ref="account:account-1",
|
|
)
|
|
|
|
self.assertEqual(2, len(second.decisions))
|
|
self.assertEqual(1, len(current_decisions(second)))
|
|
self.assertEqual("c" * 64, current_decisions(second)[0].input_hash)
|
|
self.assertNotEqual(first_fingerprint, decision_set_fingerprint(second))
|
|
payload = decision_set_payload(second)
|
|
self.assertEqual(2, len(payload["history"]))
|
|
self.assertEqual(1, len(payload["current_decisions"]))
|
|
with self.assertRaises(RevisionConflictError):
|
|
record_decision(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
decision_set_id=decision_set.id,
|
|
expected_revision=2,
|
|
key_hash="d" * 64,
|
|
input_hash="e" * 64,
|
|
action="defer",
|
|
reason="Needs another source.",
|
|
correction=None,
|
|
actor_ref="account:account-1",
|
|
)
|
|
|
|
def test_decision_set_is_a_fingerprinted_reference_source(self) -> None:
|
|
decision_set = create_decision_set(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
pipeline_id=self.pipeline.id,
|
|
name="July review",
|
|
node_id="reconcile",
|
|
actor_ref="account:account-1",
|
|
)
|
|
record_decision(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
decision_set_id=decision_set.id,
|
|
expected_revision=1,
|
|
key_hash="a" * 64,
|
|
input_hash="b" * 64,
|
|
action="reject",
|
|
reason="The observed record belongs to another case.",
|
|
correction=None,
|
|
actor_ref="account:account-1",
|
|
)
|
|
source = decision_set_source_payload(decision_set)
|
|
resolver = _datasource_source_resolver(
|
|
session=self.session,
|
|
principal=self.principal,
|
|
registry=None,
|
|
)
|
|
node = GraphNode(
|
|
id="decisions",
|
|
type="source.reference",
|
|
label="Review decisions",
|
|
position=GraphPosition(x=0, y=0),
|
|
config={
|
|
"source_ref": source["ref"],
|
|
"expected_fingerprint": source["fingerprint"],
|
|
},
|
|
)
|
|
|
|
resolved = resolver(node, 100)
|
|
|
|
self.assertEqual("dataflow.reconciliation_decisions", resolved.provider)
|
|
self.assertEqual(1, resolved.total_rows)
|
|
self.assertEqual("reject", resolved.rows[0]["action"])
|
|
stale = node.model_copy(deep=True)
|
|
stale.config["expected_fingerprint"] = "sha256:" + "0" * 64
|
|
with self.assertRaisesRegex(PipelineExecutionError, "changed"):
|
|
resolver(stale, 100)
|
|
|
|
def test_tenant_can_keep_decisions_for_visible_system_pipeline(self) -> None:
|
|
system_pipeline = DataflowPipeline(
|
|
tenant_id=None,
|
|
scope_type="system",
|
|
scope_id=None,
|
|
name="Governed monthly reconciliation",
|
|
status="active",
|
|
created_by="system-admin",
|
|
updated_by="system-admin",
|
|
)
|
|
self.session.add(system_pipeline)
|
|
self.session.flush()
|
|
|
|
decision_set = create_decision_set(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
pipeline_id=system_pipeline.id,
|
|
name="Tenant July review",
|
|
node_id=None,
|
|
actor_ref="account:account-1",
|
|
)
|
|
|
|
self.assertEqual("tenant-1", decision_set.tenant_id)
|
|
self.assertEqual(system_pipeline.id, decision_set.pipeline_id)
|
|
|
|
def test_summary_and_current_projection_do_not_require_full_history(self) -> None:
|
|
decision_set = create_decision_set(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
pipeline_id=self.pipeline.id,
|
|
name="Bounded review",
|
|
node_id="reconcile",
|
|
actor_ref="account:account-1",
|
|
)
|
|
for expected_revision, key_hash, input_hash in (
|
|
(1, "a" * 64, "b" * 64),
|
|
(2, "c" * 64, "d" * 64),
|
|
(3, "a" * 64, "e" * 64),
|
|
):
|
|
record_decision(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
decision_set_id=decision_set.id,
|
|
expected_revision=expected_revision,
|
|
key_hash=key_hash,
|
|
input_hash=input_hash,
|
|
action="accept",
|
|
reason="Reviewed against the current input.",
|
|
correction=None,
|
|
actor_ref="account:account-1",
|
|
)
|
|
|
|
summary = decision_set_payload(decision_set, include_decisions=False)
|
|
rows, total = current_decision_rows(
|
|
self.session,
|
|
decision_set_id=decision_set.id,
|
|
limit=1,
|
|
)
|
|
|
|
self.assertFalse(summary["decisions_included"])
|
|
self.assertEqual([], summary["current_decisions"])
|
|
self.assertEqual([], summary["history"])
|
|
self.assertEqual(2, total)
|
|
self.assertEqual(1, len(rows))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|