264 lines
9.5 KiB
Python
264 lines
9.5 KiB
Python
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"""<?xml version="1.0"?>
|
|
<rss version="2.0"><channel><title>Updates</title>
|
|
<link>https://example.test/</link><description>Updates</description>
|
|
<item><guid>1</guid><title>One</title></item></channel></rss>"""
|
|
|
|
|
|
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()
|