feat: make DELETE /sessions/{id} async via the deactivate command (25d6ee1e) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/router.py

+3 −0

Original line number	Diff line number	Diff line
		@@ -104,10 +104,13 @@ async def get_qod_session(
		async def delete_qod_session(
		sessionId: str,
		service: QoDService,
		caller: Caller,
		x_correlator: XCorrelatorHeader = None,
		) -> Response:
		await service.delete_session(
		session_id=sessionId,
		tenant_id=caller.tenant_id,
		app_provider_id=caller.app_provider_id,
		x_correlator=x_correlator,
		)
		return Response(status_code=status.HTTP_204_NO_CONTENT)

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+37 −4

Original line number	Diff line number	Diff line
		@@ -12,8 +12,9 @@ from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (
		)
		from open_exposure_gateway.application.mappers.quality_on_demand_mapper import (
		build_activate_command,
		build_deactivate_command,
		)
		from open_exposure_gateway.core.exceptions import DownstreamServiceException
		from open_exposure_gateway.core.exceptions import DownstreamServiceException, NotFoundException
		from open_exposure_gateway.domain.models import (
		Operation,
		OperationStatus,
		@@ -195,9 +196,41 @@ class QualityOnDemandService:
		async def delete_session(
		self,
		session_id: str,
		tenant_id: str,
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		) -> None:
		await self.srm_client.delete_qod_session(
		session_id=session_id,
		x_correlator=x_correlator,
		if self._operation_repo is None or self._qod_session_repo is None:
		raise RuntimeError("Operation/QodSession repositories are not available")

		qod_session = await self._qod_session_repo.get_by_id(UUID(session_id))
		if qod_session is None or qod_session.external_ref is None:
		raise NotFoundException(message=f"Session {session_id} not found")

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		command = build_deactivate_command(
		external_ref=qod_session.external_ref,
		operation_id=operation_id,
		correlation_id=correlation_id,
		requested_at=requested_at,
		app_provider_id=app_provider_id,
		)

		await self._operation_repo.save(
		Operation(
		operation_id=operation_id,
		correlation_id=correlation_id,
		tenant_id=tenant_id,
		app_provider_id=app_provider_id,
		operation_type=OperationType.NETWORK_CAPABILITY_DEACTIVATE,
		status=OperationStatus.PENDING,
		subject=Subject.TASK_DEACTIVATE,
		metadata={"session_id": session_id},
		)
		)

		await self._publish(
		Subject.TASK_DEACTIVATE,
		command,
		"Failed to publish QoD session deactivation",
		)

tests/unit/test_qod_flows.py

+42 −11

Original line number	Diff line number	Diff line
		@@ -16,6 +16,7 @@ from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.router import (
		from open_exposure_gateway.domain.models import OperationStatus, QodSessionState
		from open_exposure_gateway.domain.quality_on_demand import (
		SRMNetworkCapabilityActivateCommand,
		SRMNetworkCapabilityDeactivateCommand,
		Subject,
		)
		from tests.unit.fakes import (
		@@ -130,28 +131,58 @@ class TestQodSessionFlow:
		response = api_client.get(f"{QOD_BASE}/sessions/{uuid4()}")
		assert response.status_code == 404

		@pytest.mark.skip(
		reason="TODO: DELETE /sessions/{id} still calls srm_client.delete_qod_session "
		"unchanged, keyed on OEG's sessionId — same unresolved identity mapping as GET "
		"(see test_get_session_returns_created_session). Deletion 404s until that mapping "
		"exists and the deactivate command (command.srm.network.capability.deactivate) is "
		"implemented."
		)
		def test_delete_session_releases_the_downstream_reservation(
		def test_delete_session_returns_404_before_activation_confirmed(
		self, api_client: TestClient
		) -> None:
		"""DELETE must actually tear the session down at SRM: a 204 with the
		reservation still alive downstream silently leaks QoS resources."""
		"""A session with no confirmed external_ref yet (still REQUESTED) has nothing
		for a deactivate command to key on, so deletion 404s rather than silently
		no-op-ing or guessing at an id SRM never confirmed."""
		session_id = api_client.post(f"{QOD_BASE}/sessions", json=SESSION_BODY).json()["sessionId"]

		response = api_client.delete(f"{QOD_BASE}/sessions/{session_id}")

		assert response.status_code == 204
		assert response.status_code == 404

		def test_delete_session_returns_404_when_unknown(self, api_client: TestClient) -> None:
		response = api_client.delete(f"{QOD_BASE}/sessions/{uuid4()}")
		assert response.status_code == 404

		async def test_delete_session_publishes_deactivate_command_once_available(
		self,
		api_client: TestClient,
		live_qod: None,
		fake_bus: FakeDataBus,
		qod_session_repo: FakeQodSessionRepository,
		) -> None:
		"""DELETE must actually tear the session down at SRM: a 204 with no deactivate
		command published would silently leak QoS resources."""
		response = api_client.post(f"{QOD_BASE}/sessions", json=SESSION_BODY)
		session_id = UUID(response.json()["sessionId"])
		operation_id = qod_session_repo.rows[session_id].operation_id

		await fake_bus.publish(
		str(Subject.OPERATION_COMPLETED),
		completion_payload(
		str(operation_id),
		instances=[
		{
		"service_instance_id": str(uuid4()),
		"zone_id": str(uuid4()),
		"status": "completed",
		"external_ref": "qod-session-nef-123",
		}
		],
		),
		)

		delete_response = api_client.delete(f"{QOD_BASE}/sessions/{session_id}")

		assert delete_response.status_code == 204
		deactivates = [p for s, p in fake_bus.published if s == Subject.TASK_DEACTIVATE]
		assert len(deactivates) == 1
		command = SRMNetworkCapabilityDeactivateCommand.model_validate(deactivates[0])
		assert command.network_capability.external_ref == "qod-session-nef-123"


		class TestQodSessionValidation:
		"""CAMARA qod-api.yaml constraints the request schema must enforce (400s)."""

tests/unit/test_qod_service.py

+124 −2

Original line number	Diff line number	Diff line
		@@ -11,7 +11,7 @@ from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (
		from open_exposure_gateway.application.services.quality_on_demand_service import (
		QualityOnDemandService,
		)
		from open_exposure_gateway.core.exceptions import DownstreamServiceException
		from open_exposure_gateway.core.exceptions import DownstreamServiceException, NotFoundException
		from open_exposure_gateway.domain.models import (
		Operation,
		OperationStatus,
		@@ -19,7 +19,10 @@ from open_exposure_gateway.domain.models import (
		QodSession,
		QodSessionState,
		)
		from open_exposure_gateway.domain.quality_on_demand import Subject
		from open_exposure_gateway.domain.quality_on_demand import (
		SRMNetworkCapabilityDeactivateCommand,
		Subject,
		)
		from open_exposure_gateway.domain.srm_events import SRMOperationCompleted
		from tests.unit.fakes import FakeOperationRepository, FakeQodSessionRepository, completion_payload

		@@ -301,3 +304,122 @@ class TestSessionIdMinting:
		assert first.sessionId != second.sessionId
		assert len(operation_repo.rows) == 2
		assert len({str(uuid) for uuid in operation_repo.rows}) == 2


		class TestDeleteSession:
		@pytest.fixture()
		def operation_repo(self) -> FakeOperationRepository:
		return FakeOperationRepository()

		@pytest.fixture()
		def qod_session_repo(self) -> FakeQodSessionRepository:
		return FakeQodSessionRepository()

		@pytest.fixture()
		def publisher(self) -> AsyncMock:
		return AsyncMock()

		@pytest.fixture()
		def service(
		self,
		operation_repo: FakeOperationRepository,
		qod_session_repo: FakeQodSessionRepository,
		publisher: AsyncMock,
		) -> QualityOnDemandService:
		return QualityOnDemandService(
		srm_client=AsyncMock(),
		publisher=publisher,
		operation_repo=operation_repo,
		qod_session_repo=qod_session_repo,
		)

		async def _seed_available_session(
		self, qod_session_repo: FakeQodSessionRepository
		) -> QodSession:
		return await qod_session_repo.save(
		QodSession(
		session_id=uuid4(),
		operation_id=uuid4(),
		service_specification_id=uuid4(),
		qos_profile="voice",
		duration_seconds=3600,
		state=QodSessionState.AVAILABLE,
		external_ref="qod-session-nef-123",
		)
		)

		async def test_publishes_deactivate_command_keyed_on_external_ref(
		self,
		service: QualityOnDemandService,
		qod_session_repo: FakeQodSessionRepository,
		publisher: AsyncMock,
		) -> None:
		qod_session = await self._seed_available_session(qod_session_repo)

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

		publisher.publish.assert_awaited_once()
		subject, payload = publisher.publish.call_args.args
		assert subject == Subject.TASK_DEACTIVATE
		command = SRMNetworkCapabilityDeactivateCommand.model_validate(payload)
		assert command.network_capability.external_ref == "qod-session-nef-123"
		assert command.app_provider_id == "provider-1"

		async def test_persists_pending_deactivate_operation_row(
		self,
		service: QualityOnDemandService,
		qod_session_repo: FakeQodSessionRepository,
		operation_repo: FakeOperationRepository,
		) -> None:
		qod_session = await self._seed_available_session(qod_session_repo)

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

		operations = list(operation_repo.rows.values())
		assert len(operations) == 1
		assert operations[0].operation_type == OperationType.NETWORK_CAPABILITY_DEACTIVATE
		assert operations[0].status == OperationStatus.PENDING
		assert operations[0].subject == Subject.TASK_DEACTIVATE

		async def test_raises_not_found_when_session_unknown(
		self, service: QualityOnDemandService
		) -> None:
		with pytest.raises(NotFoundException):
		await service.delete_session(
		session_id=str(uuid4()), tenant_id="t", app_provider_id="p"
		)

		async def test_raises_not_found_when_external_ref_not_yet_confirmed(
		self,
		service: QualityOnDemandService,
		qod_session_repo: FakeQodSessionRepository,
		) -> None:
		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.REQUESTED,
		)
		)
		with pytest.raises(NotFoundException):
		await service.delete_session(
		session_id=str(qod_session.session_id), tenant_id="t", app_provider_id="p"
		)

		async def test_raises_when_repositories_unavailable(self) -> None:
		service = QualityOnDemandService(srm_client=AsyncMock(), publisher=AsyncMock())
		with pytest.raises(RuntimeError, match="Operation/QodSession repositories"):
		await service.delete_session(
		session_id=str(uuid4()), tenant_id="t", app_provider_id="p"
		)

Admin message