from __future__ import annotations from datetime import datetime, timedelta, timezone import unittest from unittest.mock import patch from sqlalchemy import create_engine, select from sqlalchemy.orm import Session from govoplan_connectors.backend.db.models import ConnectorTabularSource from govoplan_connectors.backend.feeds import ConnectorFeedProvider from govoplan_connectors.backend.recovery import ( CONNECTOR_RECOVERY_OPERATIONS, ConnectorRecoveryError, begin_connector_external_mutation, begin_connector_read_snapshot, ) from govoplan_connectors.backend.router import api_import_feed_snapshot from govoplan_connectors.backend.schemas import FeedImportRequest from govoplan_connectors.backend.tabular_sources import WRITE_SCOPE from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.core.recovery import ( RecoveryCheckpoint, RecoveryOperation, RecoveryStatus, ) from govoplan_core.core.recovery_runtime import ( RecoveryOperationStateConflict, claim_durable_recovery_operation, ) from govoplan_core.core.tabular_sources import TabularSnapshotInput from govoplan_core.core.runtime_coordination import ( DistributedLease, RuntimeIdentity, bind_process_runtime_identity, ) from govoplan_core.db.base import Base from govoplan_connectors.backend.tabular_sources import SqlTabularSourceProvider RSS = b""" Updates https://example.test/Updates 1One""" def _identity(node: str, incarnation: str) -> RuntimeIdentity: return RuntimeIdentity( installation_id="connector-recovery-tests", node_id=node, incarnation=incarnation, role="worker", software_version="test", composition_hash="a" * 64, ) def _principal() -> ApiPrincipal: return ApiPrincipal( principal=PrincipalRef( account_id="account-1", membership_id="membership-1", tenant_id="tenant-1", scopes=frozenset({WRITE_SCOPE}), ), account=object(), user=object(), ) class ConnectorRecoveryTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all( self.engine, tables=( DistributedLease.__table__, RecoveryOperation.__table__, RecoveryCheckpoint.__table__, ConnectorTabularSource.__table__, ), ) self.session = Session(self.engine, expire_on_commit=False) bind_process_runtime_identity(_identity("node-1", "incarnation-1")) def tearDown(self) -> None: bind_process_runtime_identity(None) self.session.close() self.engine.dispose() def test_feed_snapshot_and_recovery_checkpoint_commit_atomically_and_replay(self) -> None: document = ConnectorFeedProvider().parse( RSS, source_url="https://example.test/feed.xml", ) payload = FeedImportRequest( url="https://example.test/feed.xml", name="Updates", source_name="updates", ) with ( patch( "govoplan_connectors.backend.router.feed_transport.fetch", return_value=document, ) as fetch, patch("govoplan_connectors.backend.router.audit_event"), ): first = api_import_feed_snapshot( payload, session=self.session, principal=_principal(), idempotency_key="feed-import-1", ) replay = api_import_feed_snapshot( payload, session=self.session, principal=_principal(), idempotency_key="feed-import-1", ) self.assertEqual(first.ref, replay.ref) fetch.assert_called_once() operation = self.session.scalar(select(RecoveryOperation)) assert operation is not None self.assertEqual(RecoveryStatus.SUCCEEDED.value, operation.status) self.assertEqual( first.ref.removeprefix("snapshot:"), operation.resource_id, ) def test_recovery_metadata_distinguishes_reads_from_external_mutations(self) -> None: declarations = { item.operation_type: item for item in CONNECTOR_RECOVERY_OPERATIONS } self.assertFalse(declarations["read-snapshot"].provider_mutation) self.assertTrue(declarations["read-snapshot"].implemented) self.assertTrue(declarations["external-mutation"].provider_mutation) self.assertTrue(declarations["external-mutation"].implemented) def test_stale_atomic_connector_fence_fails_without_claiming_an_effect(self) -> None: recovery = begin_connector_read_snapshot( self.session, tenant_id="tenant-1", provider_id="provider-1", idempotency_key="read-1", source_revision="revision-1", cursor="cursor-1", dry_run_evidence={"performed": True, "approved": True}, ) lease = self.session.scalar(select(DistributedLease)) assert lease is not None lease.expires_at = datetime.now(timezone.utc) - timedelta(seconds=1) self.session.commit() bind_process_runtime_identity(_identity("node-2", "incarnation-2")) with self.assertRaises(RecoveryOperationStateConflict): claim_durable_recovery_operation( recovery.operation.session_factory, identity=_identity("node-2", "incarnation-2"), operation_id=recovery.operation_id, ) operation = self.session.get(RecoveryOperation, recovery.operation_id) self.session.refresh(operation) self.assertEqual(RecoveryStatus.FAILED.value, operation.status) def test_external_mutation_unknown_outcome_blocks_blind_retry(self) -> None: kwargs = { "tenant_id": "tenant-1", "provider_id": "provider-1", "idempotency_key": "publish-1", "request_sha256": "b" * 64, "source_revision": "revision-1", "cursor": None, "dry_run_evidence": {"performed": True, "approved": True}, "resource_type": "external_record", "resource_id": "record-1", } recovery = begin_connector_external_mutation(self.session, **kwargs) recovery.outcome_unknown( summary="The provider connection closed after dispatch", provider_code="connection_closed", ) with self.assertRaises(ConnectorRecoveryError): begin_connector_external_mutation(self.session, **kwargs) operation = self.session.get(RecoveryOperation, recovery.operation_id) self.session.refresh(operation) self.assertEqual(RecoveryStatus.OUTCOME_UNKNOWN.value, operation.status) def test_tampered_chain_rolls_back_the_atomic_connector_projection(self) -> None: recovery = begin_connector_read_snapshot( self.session, tenant_id="tenant-1", provider_id="provider-1", idempotency_key="tampered-read", source_revision=None, cursor=None, dry_run_evidence={"performed": False, "reason": "read-only"}, ) checkpoint = self.session.scalar( select(RecoveryCheckpoint) .where(RecoveryCheckpoint.operation_id == recovery.operation_id) .order_by(RecoveryCheckpoint.sequence) .limit(1) ) assert checkpoint is not None checkpoint.summary = "tampered" self.session.commit() source = SqlTabularSourceProvider().create_snapshot( self.session, _principal(), snapshot=TabularSnapshotInput( name="Tampered", source_name="tampered", rows=({"id": 1},), ), source_id=recovery.resource_id, ) with self.assertRaises(ConnectorRecoveryError): recovery.commit_success( self.session, evidence={ "verified": True, "checks": {"snapshot_ref": source.ref}, }, ) self.assertIsNone( self.session.get(ConnectorTabularSource, recovery.resource_id) ) operation = self.session.get(RecoveryOperation, recovery.operation_id) self.session.refresh(operation) self.assertEqual(RecoveryStatus.RUNNING.value, operation.status) def test_definitive_external_rejection_is_terminal(self) -> None: recovery = begin_connector_external_mutation( self.session, tenant_id="tenant-1", provider_id="provider-1", idempotency_key="publish-rejected", request_sha256="c" * 64, source_revision="revision-1", cursor=None, dry_run_evidence={"performed": True, "approved": True}, resource_type="external_record", resource_id="record-2", ) recovery.reject( summary="The provider rejected the requested revision", provider_code="revision_conflict", ) operation = self.session.get(RecoveryOperation, recovery.operation_id) self.session.refresh(operation) self.assertEqual(RecoveryStatus.REJECTED.value, operation.status) if __name__ == "__main__": unittest.main()