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()