import json from concurrent.futures import ThreadPoolExecutor from io import StringIO from unittest import mock, skipUnless from django.conf import settings from django.core.management import call_command from django.core.management.base import CommandError from django.db import close_old_connections, connection, transaction from django.test import TransactionTestCase, override_settings from api.models import ( IntegrationAggregateVersion, IntegrationOutbox, IntegrationTransportCursor, TtActivity, TtLocation, TtStaff, ) from api.services.integration.bootstrap import ( RESOURCE_PROJECTION_REVISION_ORIGIN, ResourceBootstrapSafetyError, bootstrap_resource_snapshots, revise_resource_projection_snapshots, ) from api.services.integration.outbox import ( append_resource_event, build_transport_envelope, capture_resource_state, finalize_change_set as finalize_outbox_change_set, ) BOOTSTRAP_SETTINGS = { **settings.RESOURCE_BOOKING_INTEGRATION, "CAPTURE_ENABLED": True, "PUBLISH_ENABLED": False, "ACTIVATION_APPROVED": False, "REVERSE_DELIVERY_ENABLED": False, "PHASE": "phase1", "SCHEMA_VERSION": "2", "DEPLOYMENT_ID": "bootstrap-campus", "SOURCE_SCOPE": "bootstrap-campus", "SNAPSHOT_SERVICE_TOKEN": "bootstrap-snapshot-token", "SNAPSHOT_MAX_PAGE_SIZE": 100, "SNAPSHOT_MAX_BYTES": 2097152, } @override_settings(RESOURCE_BOOKING_INTEGRATION=BOOTSTRAP_SETTINGS) class ResourceBookingResourceBootstrapTests(TransactionTestCase): reset_sequences = True def snapshot(self, **query): return self.client.get( "/api/integration/resource-booking/v1/snapshot", {"source_scope": "bootstrap-campus", **query}, HTTP_AUTHORIZATION="Bearer bootstrap-snapshot-token", ) def test_bootstrap_emits_all_existing_resources_without_fabricating_activities( self, ): staff = [ TtStaff.objects.create(code=f"S{index}", name=f"Staff {index}", status=1) for index in range(1, 4) ] archived = TtLocation.objects.create(code="L1", name="Archived Room", status=2) TtLocation.objects.create(code="L2", name="Active Room", status=1) TtActivity.objects.create( code="A1", name="Existing activity", status=1, duration=60 ) summary = bootstrap_resource_snapshots( batch_size=2, actor_id=42, run_id="bootstrap-test-1", ) self.assertEqual(summary.staff_scanned, 3) self.assertEqual(summary.staff_emitted, 3) self.assertEqual(summary.location_scanned, 2) self.assertEqual(summary.location_emitted, 2) self.assertEqual(summary.change_sets, 3) self.assertEqual(IntegrationOutbox.objects.count(), 5) self.assertFalse( IntegrationOutbox.objects.filter(aggregate_type="activity").exists() ) self.assertTrue( all( IntegrationOutbox.objects.values_list( "transaction_finalized", flat=True ) ) ) self.assertEqual( set(IntegrationOutbox.objects.values_list("origin", flat=True)), {"resource_bootstrap"}, ) self.assertEqual( set(IntegrationOutbox.objects.values_list("request_id", flat=True)), {"bootstrap-test-1"}, ) self.assertEqual( set(IntegrationOutbox.objects.values_list("event_version", flat=True)), {1}, ) first = IntegrationOutbox.objects.get( aggregate_type="staff", aggregate_id=f"bootstrap-campus:{staff[0].id}" ) current = first.payload["committed_state"]["replacement"]["current"] self.assertEqual( current["identity"], { "deployment_id": "bootstrap-campus", "aggregate_type": "staff", "source_id": str(staff[0].id), }, ) archived_event = IntegrationOutbox.objects.get( aggregate_type="location", aggregate_id=f"bootstrap-campus:{archived.id}" ) self.assertEqual( archived_event.payload["committed_state"]["replacement"]["current"][ "record_state" ], "archived", ) response = self.snapshot() self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["snapshot_watermark"], str(summary.last_sequence)) self.assertTrue(body["complete"]) self.assertEqual( sum(len(envelope["events"]) for envelope in body["envelopes"]), 5 ) self.assertTrue( all( envelope["event_id"] != envelope["source_transaction_id"] for envelope in body["envelopes"] ) ) repeated = self.snapshot(snapshot_watermark=body["snapshot_watermark"]).json() self.assertEqual( [envelope["event_id"] for envelope in body["envelopes"]], [envelope["event_id"] for envelope in repeated["envelopes"]], ) for change_set_id in IntegrationOutbox.objects.values_list( "change_set_id", flat=True ).distinct(): live = build_transport_envelope( IntegrationOutbox.objects.filter(change_set_id=change_set_id) ) self.assertEqual(live["event_id"], live["source_transaction_id"]) def test_rerun_is_noop_and_stale_direct_write_gets_one_next_version(self): staff = TtStaff.objects.create(code="S1", name="Before", status=1) first = bootstrap_resource_snapshots(batch_size=10, run_id="bootstrap-first") rerun = bootstrap_resource_snapshots(batch_size=10, run_id="bootstrap-rerun") self.assertEqual(first.staff_emitted, 1) self.assertEqual(rerun.staff_emitted, 0) self.assertEqual(rerun.staff_skipped, 1) self.assertEqual(rerun.last_sequence, first.last_sequence) self.assertEqual(IntegrationOutbox.objects.count(), 1) self.assertEqual(IntegrationTransportCursor.objects.get().last_sequence, 1) staff.name = "After direct writer" staff.save(update_fields=["name", "updated_at"]) repaired = bootstrap_resource_snapshots( batch_size=10, run_id="bootstrap-repair" ) self.assertEqual(repaired.staff_emitted, 1) versions = list( IntegrationOutbox.objects.filter(aggregate_type="staff") .order_by("event_version") .values_list("event_version", flat=True) ) self.assertEqual(versions, [1, 2]) replacement = IntegrationOutbox.objects.get(event_version=2).payload[ "committed_state" ]["replacement"] self.assertEqual(replacement["previous"]["name"], "Before") self.assertEqual(replacement["current"]["name"], "After direct writer") final_rerun = bootstrap_resource_snapshots( batch_size=10, run_id="bootstrap-final" ) self.assertEqual(final_rerun.resources_emitted, 0) self.assertEqual(IntegrationOutbox.objects.count(), 2) def test_existing_tombstone_is_preserved_and_no_resource_is_recreated(self): staff = TtStaff.objects.create(code="S1", name="Deleted", status=1) bootstrap_resource_snapshots(run_id="bootstrap-before-delete") previous = capture_resource_state(staff) staff_id = staff.id staff.delete() with transaction.atomic(): append_resource_event( TtStaff(id=staff_id), action="deleted", previous=previous, tombstone=True, origin="test_delete", ) tombstone = IntegrationOutbox.objects.order_by("-event_version").first() summary = bootstrap_resource_snapshots(run_id="bootstrap-after-delete") self.assertEqual(summary.resources_scanned, 0) self.assertEqual(IntegrationOutbox.objects.count(), 2) tombstone.refresh_from_db() self.assertTrue(tombstone.payload["committed_state"]["tombstone"]) self.assertIsNone( tombstone.payload["committed_state"]["replacement"]["current"] ) def test_batch_failure_rolls_back_events_versions_and_transport_cursor(self): TtStaff.objects.create(code="S1", name="One", status=1) TtLocation.objects.create(code="L1", name="Room", status=1) with mock.patch( "api.services.integration.bootstrap.finalize_change_set", side_effect=RuntimeError("injected finalization failure"), ): with self.assertRaisesRegex(RuntimeError, "injected finalization failure"): bootstrap_resource_snapshots(batch_size=10, run_id="bootstrap-failure") self.assertFalse(IntegrationOutbox.objects.exists()) self.assertFalse(IntegrationAggregateVersion.objects.exists()) self.assertFalse(IntegrationTransportCursor.objects.exists()) self.assertEqual(TtStaff.objects.count(), 1) self.assertEqual(TtLocation.objects.count(), 1) def test_command_requires_confirmation_and_reports_only_non_secret_counts(self): TtStaff.objects.create(code="S1", name="One", status=1) with self.assertRaisesRegex(CommandError, "confirm-resource-bootstrap"): call_command("bootstrap_resource_booking_resources") output = StringIO() call_command( "bootstrap_resource_booking_resources", confirm_resource_bootstrap=True, batch_size=10, run_id="bootstrap-command", stdout=output, ) result = json.loads(output.getvalue()) self.assertEqual(result["staff_emitted"], 1) self.assertEqual(result["resources_emitted"], 1) self.assertNotIn("bootstrap-snapshot-token", output.getvalue()) def test_bootstrap_refuses_enabled_or_reverse_delivery_modes(self): TtStaff.objects.create(code="S1", name="One", status=1) enabled = { **BOOTSTRAP_SETTINGS, "PUBLISH_ENABLED": True, "ACTIVATION_APPROVED": True, } with override_settings(RESOURCE_BOOKING_INTEGRATION=enabled): with self.assertRaisesRegex(ResourceBootstrapSafetyError, "publication"): bootstrap_resource_snapshots(run_id="unsafe-enabled") with self.assertRaisesRegex(ResourceBootstrapSafetyError, "publication"): revise_resource_projection_snapshots( operation_key="unsafe-revision-enabled" ) reverse = {**BOOTSTRAP_SETTINGS, "REVERSE_DELIVERY_ENABLED": True} with override_settings(RESOURCE_BOOKING_INTEGRATION=reverse): with self.assertRaisesRegex(ResourceBootstrapSafetyError, "reverse"): bootstrap_resource_snapshots(run_id="unsafe-reverse") with self.assertRaisesRegex(ResourceBootstrapSafetyError, "reverse"): revise_resource_projection_snapshots( operation_key="unsafe-revision-reverse" ) self.assertFalse(IntegrationOutbox.objects.exists()) def test_projection_revision_advances_resources_once_and_snapshot_is_latest_v2( self, ): staff = TtStaff.objects.create(code="S1", name="One", status=1) location = TtLocation.objects.create(code="L1", name="Room", status=2) TtActivity.objects.create( code="A1", name="Existing activity", status=1, duration=60 ) bootstrap = bootstrap_resource_snapshots( batch_size=10, run_id="projection-initial-bootstrap" ) revised = revise_resource_projection_snapshots( operation_key="rb-resource-projection-v2", batch_size=10, ) self.assertEqual(bootstrap.resources_emitted, 2) self.assertEqual(revised.resources_emitted, 2) self.assertEqual(revised.change_sets, 2) self.assertFalse( IntegrationOutbox.objects.filter(aggregate_type="activity").exists() ) revision_events = IntegrationOutbox.objects.filter( origin=RESOURCE_PROJECTION_REVISION_ORIGIN ) self.assertEqual(revision_events.count(), 2) self.assertEqual( set(revision_events.values_list("request_id", flat=True)), {"rb-resource-projection-v2"}, ) self.assertEqual( set(revision_events.values_list("event_version", flat=True)), {2} ) self.assertEqual( set(revision_events.values_list("event_type", flat=True)), {"timetabler.staff.snapshot", "timetabler.location.snapshot"}, ) self.assertEqual( revision_events.get(aggregate_type="staff").payload["committed_state"][ "replacement" ]["current"]["id"], staff.id, ) self.assertEqual( revision_events.get(aggregate_type="location").payload["committed_state"][ "replacement" ]["current"]["id"], location.id, ) staff.refresh_from_db() location.refresh_from_db() self.assertEqual((staff.code, staff.name, staff.status), ("S1", "One", 1)) self.assertEqual( (location.code, location.name, location.status), ("L1", "Room", 2) ) response = self.snapshot(snapshot_watermark=str(revised.last_sequence)) self.assertEqual(response.status_code, 200) members = [ member for envelope in response.json()["envelopes"] for member in envelope["events"] ] self.assertEqual(len(members), 2) self.assertEqual({member["event_version"] for member in members}, {2}) self.assertEqual( {member["aggregate"]["aggregate_type"] for member in members}, {"staff", "location"}, ) retry = revise_resource_projection_snapshots( operation_key="rb-resource-projection-v2", batch_size=10, ) ordinary_bootstrap = bootstrap_resource_snapshots( batch_size=10, run_id="projection-post-revision-bootstrap" ) self.assertEqual(retry.resources_emitted, 0) self.assertEqual(retry.resources_skipped, 2) self.assertEqual(ordinary_bootstrap.resources_emitted, 0) self.assertEqual(ordinary_bootstrap.resources_skipped, 2) self.assertEqual(IntegrationOutbox.objects.count(), 4) def test_projection_revision_retry_completes_partial_run_without_double_version( self, ): staff = [ TtStaff.objects.create(code=f"S{index}", name=f"Staff {index}", status=1) for index in range(1, 4) ] bootstrap_resource_snapshots(batch_size=10, run_id="partial-initial") finalize_calls = 0 def fail_second_batch(change_set_id): nonlocal finalize_calls finalize_calls += 1 if finalize_calls == 2: raise RuntimeError("injected revision finalization failure") return finalize_outbox_change_set(change_set_id) with mock.patch( "api.services.integration.bootstrap.finalize_change_set", side_effect=fail_second_batch, ): with self.assertRaisesRegex( RuntimeError, "injected revision finalization failure" ): revise_resource_projection_snapshots( operation_key="rb-resource-projection-partial", batch_size=2, ) versions_after_failure = { resource.id: IntegrationOutbox.objects.filter( aggregate_type="staff", aggregate_id=f"bootstrap-campus:{resource.id}", ).count() for resource in staff } self.assertEqual( versions_after_failure, {staff[0].id: 2, staff[1].id: 2, staff[2].id: 1} ) self.assertEqual(IntegrationTransportCursor.objects.get().last_sequence, 2) recovered = revise_resource_projection_snapshots( operation_key="rb-resource-projection-partial", batch_size=2, ) retried = revise_resource_projection_snapshots( operation_key="rb-resource-projection-partial", batch_size=2, ) self.assertEqual(recovered.resources_emitted, 1) self.assertEqual(recovered.resources_skipped, 2) self.assertEqual(retried.resources_emitted, 0) self.assertEqual(retried.resources_skipped, 3) self.assertEqual( set( IntegrationOutbox.objects.filter(aggregate_type="staff").values_list( "event_version", flat=True ) ), {1, 2}, ) for resource in staff: self.assertEqual( IntegrationOutbox.objects.filter( aggregate_type="staff", aggregate_id=f"bootstrap-campus:{resource.id}", ).count(), 2, ) self.assertEqual(IntegrationTransportCursor.objects.get().last_sequence, 3) def test_projection_revision_preserves_tombstones_and_does_not_recreate_rows(self): staff = TtStaff.objects.create(code="S1", name="Deleted", status=1) location = TtLocation.objects.create(code="L1", name="Room", status=1) bootstrap_resource_snapshots(run_id="revision-tombstone-initial") previous = capture_resource_state(staff) staff_id = staff.id staff.delete() with transaction.atomic(): append_resource_event( TtStaff(id=staff_id), action="deleted", previous=previous, tombstone=True, origin="test_delete", ) revised = revise_resource_projection_snapshots( operation_key="rb-resource-projection-tombstone" ) self.assertEqual(revised.staff_scanned, 0) self.assertEqual(revised.location_emitted, 1) self.assertFalse(TtStaff.objects.filter(id=staff_id).exists()) staff_events = IntegrationOutbox.objects.filter( aggregate_type="staff", aggregate_id=f"bootstrap-campus:{staff_id}" ) self.assertEqual(staff_events.count(), 2) tombstone = staff_events.get(event_version=2) self.assertTrue(tombstone.payload["committed_state"]["tombstone"]) self.assertFalse( staff_events.filter(origin=RESOURCE_PROJECTION_REVISION_ORIGIN).exists() ) self.assertEqual( IntegrationOutbox.objects.get( aggregate_type="location", aggregate_id=f"bootstrap-campus:{location.id}", origin=RESOURCE_PROJECTION_REVISION_ORIGIN, ).event_version, 2, ) def test_projection_revision_command_requires_confirmation_and_operation_key(self): TtStaff.objects.create(code="S1", name="One", status=1) bootstrap_resource_snapshots(run_id="revision-command-initial") with self.assertRaisesRegex( CommandError, "confirm-resource-projection-revision" ): call_command("revise_resource_booking_resource_projection") with self.assertRaisesRegex(CommandError, "operation_key"): call_command( "revise_resource_booking_resource_projection", confirm_resource_projection_revision=True, ) output = StringIO() call_command( "revise_resource_booking_resource_projection", confirm_resource_projection_revision=True, operation_key="rb-resource-projection-command", batch_size=10, stdout=output, ) result = json.loads(output.getvalue()) self.assertEqual(result["operation"], "resource_projection_revision") self.assertEqual(result["operation_key"], "rb-resource-projection-command") self.assertEqual(result["resources_emitted"], 1) @skipUnless( connection.vendor == "postgresql", "requires PostgreSQL row/advisory locks" ) def test_concurrent_reruns_emit_each_resource_once(self): TtStaff.objects.create(code="S1", name="One", status=1) def run(run_id): close_old_connections() try: return bootstrap_resource_snapshots(batch_size=10, run_id=run_id) finally: close_old_connections() with ThreadPoolExecutor(max_workers=2) as executor: summaries = list(executor.map(run, ("concurrent-1", "concurrent-2"))) self.assertEqual(sum(summary.staff_emitted for summary in summaries), 1) self.assertEqual(IntegrationOutbox.objects.count(), 1) self.assertEqual(IntegrationAggregateVersion.objects.get().version, 1) self.assertEqual(IntegrationTransportCursor.objects.get().last_sequence, 1) @skipUnless( connection.vendor == "postgresql", "requires PostgreSQL row/advisory locks" ) def test_concurrent_projection_revision_retries_advance_once(self): TtStaff.objects.create(code="S1", name="One", status=1) bootstrap_resource_snapshots( batch_size=10, run_id="concurrent-revision-initial" ) def run(): close_old_connections() try: return revise_resource_projection_snapshots( batch_size=10, operation_key="rb-resource-projection-concurrent", ) finally: close_old_connections() with ThreadPoolExecutor(max_workers=2) as executor: summaries = list(executor.map(lambda _index: run(), range(2))) self.assertEqual(sum(summary.staff_emitted for summary in summaries), 1) self.assertEqual(IntegrationOutbox.objects.count(), 2) self.assertEqual(IntegrationAggregateVersion.objects.get().version, 2) self.assertEqual(IntegrationTransportCursor.objects.get().last_sequence, 2)