import datetime import uuid from unittest import mock from django.conf import settings from django.test import TransactionTestCase, override_settings from api.models import ( IntegrationAggregateVersion, IntegrationOutbox, IntegrationTransportCursor, TtAcademicTerm, TtActivity, TtLocation, TtStaff, TtWeek, ) from api.services.integration.term_projection_backfill import ( TermProjectionBackfillContract, TermProjectionBackfillError, _sha256, build_plan, execute_backfill, ) from api.services.integration.outbox import IntegrationAtomicityError TEST_SHA = "a" * 40 class TermProjectionBackfillTests(TransactionTestCase): reset_sequences = True def setUp(self): self.term = TtAcademicTerm.objects.create( code="2026-T1", name="Term 1", start_date=datetime.date(2026, 1, 5), end_date=datetime.date(2026, 6, 26), start_time=datetime.time(8), end_time=datetime.time(18), status=1, ) self.week = TtWeek.objects.create( week=1, start_date=datetime.date(2026, 1, 5) ) self.staff = TtStaff.objects.create(code="S1", name="Staff", status=1) self.location = TtLocation.objects.create(code="L1", name="Room", status=1) first = self._activity("A1", staff=True) second = self._activity("A2", staff=False) self.activity_ids = (first.id, second.id) self.contract = TermProjectionBackfillContract( term_id=self.term.id, term_start=self.term.start_date, term_end=self.term.end_date, activity_ids=self.activity_ids, activity_ids_sha256=_sha256(list(self.activity_ids)), entry_watermark=0, final_watermark=1, expected_occurrence_count=2, expected_missing_staff_ids=(second.id,), expected_database_configuration_sha256=None, expected_current_database_sha256=None, operation_key="test-term-source-backfill-v1", operation_origin="test_term_source_backfill", change_set_id=uuid.uuid5(uuid.NAMESPACE_URL, "test-term-source-backfill-v1"), ) IntegrationTransportCursor.objects.create(source_scope="default", last_sequence=0) def _activity(self, code: str, *, staff: bool) -> TtActivity: activity = TtActivity.objects.create( code=code, name=code, academic_term=self.term, status=1, duration=60, scheduled=1, scheduled_day=0, scheduled_start_time=datetime.time(9), scheduled_start_slot=9, ) activity.week.add(self.week) activity.location.add(self.location) if staff: activity.staff.add(self.staff) return activity def _integration_settings(self, **changes): return { **settings.RESOURCE_BOOKING_INTEGRATION, "CAPTURE_ENABLED": True, "PUBLISH_ENABLED": True, "ACTIVATION_APPROVED": True, "PHASE": "phase1", "REVERSE_DELIVERY_ENABLED": False, "TRANSPORT": "kafka", "SCHEMA_VERSION": "2", "SOURCE_SCOPE": "default", "DEPLOYMENT_ID": "default", "MAX_EVENT_BYTES": 2_097_152, **changes, } def _readiness(self, watermark): return { "transport_watermark": watermark, "publisher_last_sequence": watermark, "publisher_live": True, "publisher_status": "running", "dead_letter_count": 0, "outbox_backlog": 0, } def test_plan_and_execute_one_atomic_truthful_location_only_change_set(self): before = list( TtActivity.objects.filter(id__in=self.activity_ids) .order_by("id") .values("id", "scheduled", "scheduled_day", "scheduled_start_slot") ) with override_settings( RESOURCE_BOOKING_INTEGRATION=self._integration_settings() ), mock.patch( "api.services.integration.term_projection_backfill._assert_readiness", side_effect=lambda watermark: self._readiness(watermark), ): plan = build_plan( deployed_git_sha=TEST_SHA, contract=self.contract, ) evidence = execute_backfill( deployed_git_sha=TEST_SHA, expected_plan_sha256=plan["plan_sha256"], contract=self.contract, ) retry = execute_backfill( deployed_git_sha=TEST_SHA, expected_plan_sha256=plan["plan_sha256"], contract=self.contract, ) rows = list(IntegrationOutbox.objects.order_by("transaction_index")) after = list( TtActivity.objects.filter(id__in=self.activity_ids) .order_by("id") .values("id", "scheduled", "scheduled_day", "scheduled_start_slot") ) self.assertEqual(before, after) self.assertEqual(len(rows), 2) self.assertEqual({row.transport_sequence for row in rows}, {1}) self.assertEqual({row.change_set_id for row in rows}, {self.contract.change_set_id}) self.assertTrue(all(row.transaction_finalized for row in rows)) self.assertEqual([row.transaction_count for row in rows], [2, 2]) self.assertEqual(plan["truthful_missing_staff_activity_ids"], [self.activity_ids[1]]) self.assertEqual(evidence["truthful_missing_staff_activity_ids"], [self.activity_ids[1]]) self.assertEqual(evidence["absolute_occurrence_count"], 2) self.assertEqual(evidence["canonical_bytes"], plan["transport_preview"]["canonical_bytes"]) self.assertFalse(evidence["domain_mutation_executed"]) self.assertTrue(retry["idempotent_retry"]) self.assertEqual(IntegrationOutbox.objects.count(), 2) def test_state_drift_blocks_before_outbox_write(self): with override_settings( RESOURCE_BOOKING_INTEGRATION=self._integration_settings() ), mock.patch( "api.services.integration.term_projection_backfill._assert_readiness", side_effect=lambda watermark: self._readiness(watermark), ): plan = build_plan( deployed_git_sha=TEST_SHA, contract=self.contract, ) TtActivity.objects.filter(pk=self.activity_ids[0]).update(scheduled=0) with self.assertRaisesRegex( TermProjectionBackfillError, "scheduled state changed" ): execute_backfill( deployed_git_sha=TEST_SHA, expected_plan_sha256=plan["plan_sha256"], contract=self.contract, ) self.assertFalse(IntegrationOutbox.objects.exists()) def test_composite_size_failure_rolls_back_all_members_versions_and_cursor(self): with override_settings( RESOURCE_BOOKING_INTEGRATION=self._integration_settings() ), mock.patch( "api.services.integration.term_projection_backfill._assert_readiness", side_effect=lambda watermark: self._readiness(watermark), ): plan = build_plan( deployed_git_sha=TEST_SHA, contract=self.contract, ) constrained = self._integration_settings( MAX_EVENT_BYTES=plan["transport_preview"]["canonical_bytes"] - 1 ) with override_settings( RESOURCE_BOOKING_INTEGRATION=constrained ), mock.patch( "api.services.integration.term_projection_backfill._assert_readiness", side_effect=lambda watermark: self._readiness(watermark), ), mock.patch( "api.services.integration.term_projection_backfill.build_plan", return_value=plan, ): with self.assertRaisesRegex( IntegrationAtomicityError, "configured transport byte limit" ): execute_backfill( deployed_git_sha=TEST_SHA, expected_plan_sha256=plan["plan_sha256"], contract=self.contract, ) self.assertFalse(IntegrationOutbox.objects.exists()) self.assertFalse(IntegrationAggregateVersion.objects.exists()) self.assertEqual( IntegrationTransportCursor.objects.get(source_scope="default").last_sequence, 0, ) def test_exact_default_contract_preserves_audit_facts(self): contract = TermProjectionBackfillContract() self.assertEqual(contract.term_id, 26) self.assertEqual(contract.activity_ids, tuple(range(672, 717))) self.assertEqual( contract.activity_ids_sha256, "1b599dcd8709fcdb29b292b110ef1e5cb011929ba371011ef1ae055c9e408453", ) self.assertEqual(contract.expected_missing_staff_ids, (686, 690)) self.assertEqual((contract.entry_watermark, contract.final_watermark), (2639, 2640))