fix: update QoD session handling to use service_instance_id and manage terminal states (81e460e3) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -113,7 +113,7 @@ def build_session_info(qod_session: QodSession, capability: SRMNetworkCapability


		def build_deactivate_command(
		external_ref: str,
		service_instance_id: UUID,
		operation_id: UUID,
		correlation_id: str,
		requested_at: str,
		@@ -126,7 +126,7 @@ def build_deactivate_command(
		requested_at=requested_at,
		app_provider_id=app_provider_id,
		network_capability=NetworkCapabilityDeactivateTarget(
		external_ref=external_ref,
		service_instance_id=str(service_instance_id),
		grace_period_seconds=grace_period_seconds,
		),
		)

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+42 −8

Original line number	Diff line number	Diff line
		@@ -50,6 +50,10 @@ _OPERATION_COMPLETION_STATUS_MAP: dict[str, OperationStatus] = {
		"failed": OperationStatus.FAILED,
		}

		_TERMINAL_STATES = frozenset(
		{QodSessionState.DELETION_REQUESTED, QodSessionState.DELETED, QodSessionState.ERROR}
		)


		class QualityOnDemandService:
		def __init__(
		@@ -239,11 +243,10 @@ class QualityOnDemandService:
		"operation_completed_for_unknown_operation", operation_id=event.operation_id
		)
		return

		# Not ours: event.srm.operation.completed is shared across every domain
		# (deploy/terminate and network-capability activate/deactivate all land
		# here). operation_type is how a domain recognizes its own operations.
		if operation.operation_type != OperationType.NETWORK_CAPABILITY:
		if operation.operation_type not in (
		OperationType.NETWORK_CAPABILITY,
		OperationType.NETWORK_CAPABILITY_DEACTIVATE,
		):
		return

		status = _OPERATION_COMPLETION_STATUS_MAP[event.status]
		@@ -261,6 +264,10 @@ class QualityOnDemandService:
		)
		)

		if operation.operation_type == OperationType.NETWORK_CAPABILITY_DEACTIVATE:
		await self._handle_deactivate_completed(operation, status)
		return

		qod_session = await self._qod_session_repo.get_by_operation_id(operation_id)
		if qod_session is None:
		logger.warning(
		@@ -288,6 +295,29 @@ class QualityOnDemandService:
		occurred_at=event.completed_at,
		)

		async def _handle_deactivate_completed(
		self, operation: Operation, status: OperationStatus
		) -> None:
		assert self._qod_session_repo is not None
		session_id_raw = (operation.metadata or {}).get("session_id")
		if session_id_raw is None:
		logger.warning(
		"deactivate_completed_without_session_id", operation_id=str(operation.operation_id)
		)
		return

		qod_session = await self._qod_session_repo.get_by_id(UUID(session_id_raw))
		if qod_session is None:
		logger.warning(
		"deactivate_completed_for_unknown_session", session_id=session_id_raw
		)
		return

		terminal_state = (
		QodSessionState.ERROR if status == OperationStatus.FAILED else QodSessionState.DELETED
		)
		await self._qod_session_repo.save(qod_session.model_copy(update={"state": terminal_state}))

		async def handle_status_changed(self, event: SRMOperationStatus) -> None:
		if self._operation_repo is None or self._qod_session_repo is None:
		raise RuntimeError("Operation/QodSession repositories are not available")
		@@ -332,7 +362,7 @@ class QualityOnDemandService:
		raise RuntimeError("QodSession repository is not available")

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

		capability = await self.srm_client.get_network_capability(
		@@ -352,12 +382,12 @@ class QualityOnDemandService:
		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:
		if qod_session 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,
		service_instance_id=qod_session.session_id,
		operation_id=operation_id,
		correlation_id=correlation_id,
		requested_at=requested_at,
		@@ -383,6 +413,10 @@ class QualityOnDemandService:
		"Failed to publish QoD session deactivation",
		)

		await self._qod_session_repo.save(
		qod_session.model_copy(update={"state": QodSessionState.DELETION_REQUESTED})
		)

		if qod_session.state == QodSessionState.AVAILABLE:
		await self._deliver_qos_status_changed(
		operation_id=qod_session.operation_id,

src/open_exposure_gateway/domain/models/qod_sessions/enums.py

+3 −0

Original line number	Diff line number	Diff line
		@@ -5,3 +5,6 @@ class QodSessionState(StrEnum):
		REQUESTED = "REQUESTED"
		AVAILABLE = "AVAILABLE"
		UNAVAILABLE = "UNAVAILABLE"
		DELETION_REQUESTED = "DELETION_REQUESTED"
		DELETED = "DELETED"
		ERROR = "ERROR"

src/open_exposure_gateway/domain/quality_on_demand.py

+2 −1

Original line number	Diff line number	Diff line
		@@ -60,7 +60,8 @@ class SRMNetworkCapabilityActivateCommand(BaseModel):

		class NetworkCapabilityDeactivateTarget(BaseModel):
		capability_type: str = "qod_session"
		external_ref: str
		service_instance_id: str \| None = None
		external_ref: str \| None = None
		grace_period_seconds: int = 0

tests/unit/test_qod_flows.py

+75 −6

Original line number	Diff line number	Diff line
		@@ -182,16 +182,27 @@ class TestQodSessionFlow:
		response = api_client.get(f"{QOD_BASE}/sessions/{uuid4()}")
		assert response.status_code == 404

		def test_delete_session_returns_404_before_activation_confirmed(
		self, api_client: TestClient
		def test_delete_session_succeeds_before_activation_confirmed(
		self, api_client: TestClient, fake_bus: FakeDataBus
		) -> None:
		"""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
		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.service_instance_id == session_id

		def test_get_session_returns_404_after_delete_requested(
		self, api_client: TestClient
		) -> None:
		session_id = api_client.post(f"{QOD_BASE}/sessions", json=SESSION_BODY).json()["sessionId"]
		api_client.delete(f"{QOD_BASE}/sessions/{session_id}")

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

		assert response.status_code == 404

		def test_delete_session_returns_404_when_unknown(self, api_client: TestClient) -> None:
		@@ -232,7 +243,65 @@ class TestQodSessionFlow:
		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"
		assert command.network_capability.service_instance_id == str(session_id)

		async def test_deactivate_completion_drives_session_to_deleted(
		self,
		api_client: TestClient,
		live_qod: None,
		fake_bus: FakeDataBus,
		qod_session_repo: FakeQodSessionRepository,
		operation_repo: FakeOperationRepository,
		) -> None:
		session_id = UUID(
		api_client.post(f"{QOD_BASE}/sessions", json=SESSION_BODY).json()["sessionId"]
		)
		api_client.delete(f"{QOD_BASE}/sessions/{session_id}")
		assert qod_session_repo.rows[session_id].state == QodSessionState.DELETION_REQUESTED

		deactivate_operation_id = next(
		op.operation_id
		for op in operation_repo.rows.values()
		if op.metadata == {"session_id": str(session_id)}
		)

		await fake_bus.publish(
		str(Subject.OPERATION_COMPLETED),
		completion_payload(str(deactivate_operation_id)),
		)

		assert qod_session_repo.rows[session_id].state == QodSessionState.DELETED

		async def test_failed_deactivate_completion_drives_session_to_error(
		self,
		api_client: TestClient,
		live_qod: None,
		fake_bus: FakeDataBus,
		qod_session_repo: FakeQodSessionRepository,
		operation_repo: FakeOperationRepository,
		) -> None:
		session_id = UUID(
		api_client.post(f"{QOD_BASE}/sessions", json=SESSION_BODY).json()["sessionId"]
		)
		api_client.delete(f"{QOD_BASE}/sessions/{session_id}")

		deactivate_operation_id = next(
		op.operation_id
		for op in operation_repo.rows.values()
		if op.metadata == {"session_id": str(session_id)}
		)

		await fake_bus.publish(
		str(Subject.OPERATION_COMPLETED),
		completion_payload(
		str(deactivate_operation_id),
		status="failed",
		instances=[],
		error={"status": 500, "code": "INTERNAL", "message": "backend rejected"},
		),
		)

		assert qod_session_repo.rows[session_id].state == QodSessionState.ERROR


		class TestQosProfiles:

Admin message