fix: record a skipped calleback delivery when the sink token can't be used (4cca6f4c) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/edge_application_management_service.py

+20 −0

Changes for src/open_exposure_gateway/application/services/edge_application_management_service.py: 20 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -1313,12 +1313,14 @@ class EdgeApplicationManagementService:
		"callback_skipped_credential_expired",
		callback_registration_id=str(registration.id),
		)
		await self._record_skipped_delivery(registration.id, operation_id, "credential_expired")
		return
		except CallbackCredentialRevokedError:
		logger.info(
		"callback_skipped_credential_revoked",
		callback_registration_id=str(registration.id),
		)
		await self._record_skipped_delivery(registration.id, operation_id, "credential_revoked")
		return
		except CallbackSecretError as exc:
		logger.error(
		@@ -1326,6 +1328,9 @@ class EdgeApplicationManagementService:
		callback_registration_id=str(registration.id),
		error=str(exc),
		)
		await self._record_skipped_delivery(
		registration.id, operation_id, "credential_unusable"
		)
		return

		app_id = await self._resolve_app_id(group[0].app_registration_id)
		@@ -1373,6 +1378,21 @@ class EdgeApplicationManagementService:
		)
		)

		async def _record_skipped_delivery(
		self, callback_registration_id: UUID, operation_id: UUID, reason: str
		) -> None:
		assert self._callback_delivery_repo is not None
		await self._callback_delivery_repo.save(
		CallbackDelivery(
		id=uuid4(),
		callback_registration_id=callback_registration_id,
		operation_id=operation_id,
		attempt=1,
		state="skipped",
		last_error=reason,
		)
		)

		async def _forget_credential_after_final_notification(
		self, callback_registration_id: UUID, creating_operation_id: UUID
		) -> None:

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+20 −0

Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 20 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -283,12 +283,14 @@ class QualityOnDemandService:
		"qod_callback_skipped_credential_expired",
		callback_registration_id=str(registration.id),
		)
		await self._record_skipped_delivery(registration.id, operation_id, "credential_expired")
		return
		except CallbackCredentialRevokedError:
		logger.info(
		"qod_callback_skipped_credential_revoked",
		callback_registration_id=str(registration.id),
		)
		await self._record_skipped_delivery(registration.id, operation_id, "credential_revoked")
		return
		except CallbackSecretError as exc:
		logger.error(
		@@ -296,6 +298,9 @@ class QualityOnDemandService:
		callback_registration_id=str(registration.id),
		error=str(exc),
		)
		await self._record_skipped_delivery(
		registration.id, operation_id, "credential_unusable"
		)
		return

		source = f"{get_settings().public_base_url}/qod/v0/sessions/{session_id}"
		@@ -332,6 +337,21 @@ class QualityOnDemandService:
		)
		)

		async def _record_skipped_delivery(
		self, callback_registration_id: UUID, operation_id: UUID, reason: str
		) -> None:
		assert self._callback_delivery_repo is not None
		await self._callback_delivery_repo.save(
		CallbackDelivery(
		id=uuid4(),
		callback_registration_id=callback_registration_id,
		operation_id=operation_id,
		attempt=1,
		state="skipped",
		last_error=reason,
		)
		)

		async def handle_completed(self, event: SRMOperationCompleted) -> None:
		if self._operation_repo is None or self._qod_session_repo is None:
		raise RuntimeError("Operation/QodSession repositories are not available")

tests/integration/test_postgres.py

+19 −0

Changes for tests/integration/test_postgres.py: 19 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -495,6 +495,25 @@ async def test_callback_delivery_repo_persists_and_lists(db_session: AsyncSessio
		assert deliveries[0].state == "delivered"


		async def test_callback_delivery_repo_persists_skipped_row(db_session: AsyncSession) -> None:
		operation = await SqlOperationRepository(db_session).save(_operation())
		callback_registration = await SqlCallbackRegistrationRepository(db_session).save(
		_callback_registration(operation_id=operation.operation_id)
		)
		repo = SqlCallbackDeliveryRepository(db_session)
		delivery = _callback_delivery(
		callback_registration_id=callback_registration.id,
		operation_id=operation.operation_id,
		).model_copy(update={"state": "skipped", "last_error": "credential_expired"})

		await repo.save(delivery)
		deliveries = await repo.list_by_callback_registration_id(callback_registration.id)

		assert len(deliveries) == 1
		assert deliveries[0].state == "skipped"
		assert deliveries[0].last_error == "credential_expired"


		async def test_qod_session_repo_persists_and_loads(db_session: AsyncSession) -> None:
		operation = await SqlOperationRepository(db_session).save(_operation())
		repo = SqlQodSessionRepository(db_session)

tests/unit/test_eam_service.py

+49 −10

Changes for tests/unit/test_eam_service.py: 49 added lines, 10 removed lines.
Original line number	Diff line number	Diff line
		@@ -3458,15 +3458,33 @@ class TestHandleCompleted:
		)

		@pytest.mark.parametrize(
		("stored", "key", "expected_tokens"),
		("stored", "key", "expected_tokens", "skip_reason"),
		[
		pytest.param(None, CALLBACK_KEY, [None], id="no-credential-sends-without-auth"),
		pytest.param(None, CALLBACK_KEY, [None], None, id="no-credential-sends-without-auth"),
		pytest.param(
		timedelta(hours=1), CALLBACK_KEY, [SINK_TOKEN], id="valid-credential-sends-token"
		timedelta(hours=1),
		CALLBACK_KEY,
		[SINK_TOKEN],
		None,
		id="valid-credential-sends-token",
		),
		pytest.param(
		timedelta(seconds=-1),
		CALLBACK_KEY,
		[],
		"credential_expired",
		id="expired-credential-skips",
		),
		pytest.param(
		timedelta(hours=1),
		OTHER_CALLBACK_KEY,
		[],
		"credential_unusable",
		id="undecryptable-skips",
		),
		pytest.param(
		timedelta(hours=1), None, [], "credential_unusable", id="missing-key-skips"
		),
		pytest.param(timedelta(seconds=-1), CALLBACK_KEY, [], id="expired-credential-skips"),
		pytest.param(timedelta(hours=1), OTHER_CALLBACK_KEY, [], id="undecryptable-skips"),
		pytest.param(timedelta(hours=1), None, [], id="missing-key-skips"),
		],
		)
		async def test_delivery_uses_stored_bearer_token(
		@@ -3480,6 +3498,7 @@ class TestHandleCompleted:
		stored: timedelta \| None,
		key: bytes \| None,
		expected_tokens: list[str \| None],
		skip_reason: str \| None,
		) -> None:
		callback_credential_repo = FakeCallbackCredentialRepository()
		service = self._service_with_credentials(
		@@ -3510,7 +3529,17 @@ class TestHandleCompleted:
		await service.handle_completed(self._completed_event())

		assert callback_delivery_port.bearer_tokens == expected_tokens
		assert len(callback_delivery_repo.rows) == len(expected_tokens)
		rows = list(callback_delivery_repo.rows.values())
		assert len(rows) == 1
		if skip_reason is None:
		assert rows[0].state == "delivered"
		assert rows[0].last_error is None
		else:
		assert rows[0].state == "skipped"
		assert rows[0].last_error == skip_reason
		assert rows[0].attempt == 1
		assert rows[0].callback_registration_id == registration.id
		assert rows[0].operation_id == self.OPERATION_ID
		operation = await operation_repo.get_by_id(self.OPERATION_ID)
		assert operation is not None
		assert operation.status == OperationStatus.COMPLETED
		@@ -3664,7 +3693,10 @@ class TestHandleCompleted:
		await service.handle_completed(self._completed_event())

		assert callback_delivery_port.delivered == []
		assert callback_delivery_repo.rows == {}
		rows = list(callback_delivery_repo.rows.values())
		assert [(row.state, row.last_error) for row in rows] == [
		("skipped", "credential_expired" if revoke == "purge" else "credential_revoked")
		]
		assert [log["event"] for log in logs if log["event"].startswith("callback_skipped_")] == [
		"callback_skipped_credential_expired"
		if revoke == "purge"
		@@ -3707,10 +3739,17 @@ class TestHandleCompleted:
		terminated = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert terminated is not None

		await service._deliver_callbacks(uuid4(), "2026-07-04T10:05:00+00:00", [terminated])
		later_operation_id = uuid4()
		await service._deliver_callbacks(
		later_operation_id, "2026-07-04T10:05:00+00:00", [terminated]
		)

		assert callback_delivery_port.bearer_tokens == [SINK_TOKEN]
		assert len(callback_delivery_repo.rows) == 1
		skipped = [row for row in callback_delivery_repo.rows.values() if row.state == "skipped"]
		assert len(callback_delivery_repo.rows) == 2
		assert [(row.operation_id, row.last_error) for row in skipped] == [
		(later_operation_id, "credential_revoked")
		]

		async def test_stale_total_failure_does_not_overwrite_terminated_instance(
		self,

tests/unit/test_qod_service.py

+38 −8

Changes for tests/unit/test_qod_service.py: 38 added lines, 8 removed lines.
Original line number	Diff line number	Diff line
		@@ -708,15 +708,28 @@ class TestHandleCompleted:
		assert deliveries[0].attempt == 1

		@pytest.mark.parametrize(
		("stored", "key", "expected_tokens"),
		("stored", "key", "expected_tokens", "skip_reason"),
		[
		pytest.param(False, CALLBACK_KEY, [None], id="no-credential-sends-without-auth"),
		pytest.param(True, CALLBACK_KEY, [RAW_TOKEN], id="credential-sends-token"),
		pytest.param(False, CALLBACK_KEY, [None], None, id="no-credential-sends-without-auth"),
		pytest.param(True, CALLBACK_KEY, [RAW_TOKEN], None, id="credential-sends-token"),
		pytest.param(
		"live", CALLBACK_KEY, [RAW_TOKEN], id="credential-within-session-sends-token"
		"live",
		CALLBACK_KEY,
		[RAW_TOKEN],
		None,
		id="credential-within-session-sends-token",
		),
		pytest.param(True, OTHER_CALLBACK_KEY, [], id="undecryptable-skips"),
		pytest.param(True, None, [], id="missing-key-skips"),
		pytest.param(
		"expired",
		CALLBACK_KEY,
		[],
		"credential_expired",
		id="expired-credential-skips",
		),
		pytest.param(
		True, OTHER_CALLBACK_KEY, [], "credential_unusable", id="undecryptable-skips"
		),
		pytest.param(True, None, [], "credential_unusable", id="missing-key-skips"),
		],
		)
		async def test_delivery_uses_stored_bearer_token(
		@@ -726,6 +739,7 @@ class TestHandleCompleted:
		stored: bool \| str,
		key: bytes \| None,
		expected_tokens: list[str \| None],
		skip_reason: str \| None,
		) -> None:
		callback_registration_repo = FakeCallbackRegistrationRepository()
		callback_delivery_repo = FakeCallbackDeliveryRepository()
		@@ -762,6 +776,8 @@ class TestHandleCompleted:
		+ timedelta(seconds=3600)
		+ NOTIFICATION_AUTH_TOKEN_GRACE
		if stored == "live"
		else datetime.now(timezone.utc) - timedelta(seconds=1)
		if stored == "expired"
		else None
		),
		)
		@@ -784,7 +800,17 @@ class TestHandleCompleted:
		await service.handle_completed(event)

		assert callback_delivery_port.bearer_tokens == expected_tokens
		assert len(callback_delivery_repo.rows) == len(expected_tokens)
		rows = list(callback_delivery_repo.rows.values())
		assert len(rows) == 1
		if skip_reason is None:
		assert rows[0].state == "delivered"
		assert rows[0].last_error is None
		else:
		assert rows[0].state == "skipped"
		assert rows[0].last_error == skip_reason
		assert rows[0].attempt == 1
		assert rows[0].callback_registration_id == registration.id
		assert rows[0].operation_id == operation.operation_id
		updated_session = await qod_session_repo.get_by_id(qod_session.session_id)
		assert updated_session is not None
		assert updated_session.state == QodSessionState.AVAILABLE
		@@ -1775,13 +1801,14 @@ class TestDeleteSession:
		callback_registration_repo = FakeCallbackRegistrationRepository()
		callback_credential_repo = FakeCallbackCredentialRepository()
		callback_delivery_port = FakeQodCallbackDeliveryPort()
		callback_delivery_repo = FakeCallbackDeliveryRepository()
		service = QualityOnDemandService(
		srm_client=AsyncMock(),
		publisher=publisher,
		operation_repo=operation_repo,
		qod_session_repo=qod_session_repo,
		callback_registration_repo=callback_registration_repo,
		callback_delivery_repo=FakeCallbackDeliveryRepository(),
		callback_delivery_repo=callback_delivery_repo,
		callback_delivery_port=callback_delivery_port,
		callback_credential_repo=callback_credential_repo,
		callback_secret_key=CALLBACK_KEY,
		@@ -1815,6 +1842,9 @@ class TestDeleteSession:
		)

		assert callback_delivery_port.delivered == []
		assert [(row.state, row.last_error) for row in callback_delivery_repo.rows.values()] == [
		("skipped", "credential_expired")
		]


		class TestGetSession: