feat: send DELETE_REQUESTED notification when deleting an available QoD session (e98bf7f5) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+9 −0

Original line number	Diff line number	Diff line
		@@ -364,3 +364,12 @@ class QualityOnDemandService:
		command,
		"Failed to publish QoD session deactivation",
		)

		if qod_session.state == QodSessionState.AVAILABLE:
		await self._deliver_qos_status_changed(
		operation_id=qod_session.operation_id,
		session_id=qod_session.session_id,
		qos_status="UNAVAILABLE",
		status_info="DELETE_REQUESTED",
		occurred_at=requested_at,
		)

tests/unit/test_qod_service.py

+91 −0

Original line number	Diff line number	Diff line
		@@ -916,3 +916,94 @@ class TestDeleteSession:
		await service.delete_session(
		session_id=str(uuid4()), tenant_id="t", app_provider_id="p"
		)

		async def test_delivers_delete_requested_notification_when_session_available(
		self,
		operation_repo: FakeOperationRepository,
		qod_session_repo: FakeQodSessionRepository,
		publisher: AsyncMock,
		) -> None:
		callback_registration_repo = FakeCallbackRegistrationRepository()
		callback_delivery_repo = FakeCallbackDeliveryRepository()
		callback_delivery_port = FakeQodCallbackDeliveryPort()
		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=callback_delivery_repo,
		callback_delivery_port=callback_delivery_port,
		)
		qod_session = await self._seed_available_session(qod_session_repo)
		await callback_registration_repo.save(
		CallbackRegistration(
		id=uuid4(),
		operation_id=qod_session.operation_id,
		tenant_id="tenant-1",
		api_family="quality-on-demand",
		sink="https://client.example.com/cb",
		event_types=["org.camaraproject.qod.v0.qos-status-changed"],
		)
		)

		await service.delete_session(
		session_id=str(qod_session.session_id),
		tenant_id="tenant-1",
		app_provider_id="provider-1",
		)

		assert len(callback_delivery_port.delivered) == 1
		sink, cloud_event = callback_delivery_port.delivered[0]
		assert sink == "https://client.example.com/cb"
		assert cloud_event.data.sessionId == qod_session.session_id
		assert cloud_event.data.qosStatus == "UNAVAILABLE"
		assert cloud_event.data.statusInfo == "DELETE_REQUESTED"

		async def test_no_notification_when_session_already_unavailable(
		self,
		operation_repo: FakeOperationRepository,
		qod_session_repo: FakeQodSessionRepository,
		publisher: AsyncMock,
		) -> None:
		callback_registration_repo = FakeCallbackRegistrationRepository()
		callback_delivery_repo = FakeCallbackDeliveryRepository()
		callback_delivery_port = FakeQodCallbackDeliveryPort()
		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=callback_delivery_repo,
		callback_delivery_port=callback_delivery_port,
		)
		qod_session = await qod_session_repo.save(
		QodSession(
		session_id=uuid4(),
		operation_id=uuid4(),
		service_specification_id=uuid4(),
		qos_profile="voice",
		duration_seconds=3600,
		state=QodSessionState.UNAVAILABLE,
		external_ref="qod-session-nef-123",
		)
		)
		await callback_registration_repo.save(
		CallbackRegistration(
		id=uuid4(),
		operation_id=qod_session.operation_id,
		tenant_id="tenant-1",
		api_family="quality-on-demand",
		sink="https://client.example.com/cb",
		event_types=["org.camaraproject.qod.v0.qos-status-changed"],
		)
		)

		await service.delete_session(
		session_id=str(qod_session.session_id),
		tenant_id="tenant-1",
		app_provider_id="provider-1",
		)

		assert callback_delivery_port.delivered == []

Admin message