feat: wire QOS_STATUS_CHANGED delivery into QualityOnDemandService (c88e8c2f) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+132 −2

Original line number	Diff line number	Diff line
		from datetime import datetime, timezone
		from typing import Optional
		from typing import Literal, Optional
		from uuid import UUID, uuid4

		import structlog
		@@ -13,9 +13,13 @@ 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,
		build_qos_status_changed_event,
		)
		from open_exposure_gateway.core.config import get_settings
		from open_exposure_gateway.core.exceptions import DownstreamServiceException, NotFoundException
		from open_exposure_gateway.domain.models import (
		CallbackDelivery,
		CallbackRegistration,
		Operation,
		OperationStatus,
		OperationType,
		@@ -23,10 +27,15 @@ from open_exposure_gateway.domain.models import (
		QodSessionState,
		)
		from open_exposure_gateway.domain.quality_on_demand import Subject
		from open_exposure_gateway.domain.srm_events import SRMOperationCompleted
		from open_exposure_gateway.domain.srm_events import SRMOperationCompleted, SRMOperationStatus
		from open_exposure_gateway.ports.database.callbacks import (
		CallbackDeliveryRepository,
		CallbackRegistrationRepository,
		)
		from open_exposure_gateway.ports.database.operations import OperationRepository
		from open_exposure_gateway.ports.database.qod_sessions import QodSessionRepository
		from open_exposure_gateway.ports.databus_port import DataBusPort
		from open_exposure_gateway.ports.qod_callback_port import QodCallbackDeliveryPort
		from open_exposure_gateway.ports.srm_port import SRMClientPort

		logger: structlog.BoundLogger = structlog.get_logger(__name__)
		@@ -45,11 +54,17 @@ class QualityOnDemandService:
		publisher: Optional[DataBusPort] = None,
		operation_repo: Optional[OperationRepository] = None,
		qod_session_repo: Optional[QodSessionRepository] = None,
		callback_registration_repo: Optional[CallbackRegistrationRepository] = None,
		callback_delivery_repo: Optional[CallbackDeliveryRepository] = None,
		callback_delivery_port: Optional[QodCallbackDeliveryPort] = None,
		) -> None:
		self.srm_client = srm_client
		self._publisher = publisher
		self._operation_repo = operation_repo
		self._qod_session_repo = qod_session_repo
		self._callback_registration_repo = callback_registration_repo
		self._callback_delivery_repo = callback_delivery_repo
		self._callback_delivery_port = callback_delivery_port

		def _new_operation_metadata(self, x_correlator: Optional[str]) -> tuple[UUID, str, str]:
		operation_id = uuid4()
		@@ -114,6 +129,26 @@ class QualityOnDemandService:
		)
		)

		if request.webhook is not None:
		if self._callback_registration_repo is None:
		raise RuntimeError("CallbackRegistrationRepository is not available")
		await self._callback_registration_repo.save(
		CallbackRegistration(
		id=uuid4(),
		operation_id=operation_id,
		tenant_id=tenant_id,
		api_family="quality-on-demand",
		sink=request.webhook.notificationUrl,
		event_types=["org.camaraproject.qod.v0.qos-status-changed"],
		sink_credential_ref=(
		f"secret://oeg/{operation_id}/notification-auth-token"
		if request.webhook.notificationAuthToken
		else None
		),
		expires_at=None,
		)
		)

		await self._publish(
		Subject.TASK_ACTIVATE,
		command,
		@@ -135,6 +170,54 @@ class QualityOnDemandService:
		qosStatus=QosStatus.REQUESTED,
		)

		async def _deliver_qos_status_changed(
		self,
		operation_id: UUID,
		session_id: UUID,
		qos_status: Literal["AVAILABLE", "UNAVAILABLE"],
		status_info: Literal["DURATION_EXPIRED", "NETWORK_TERMINATED", "DELETE_REQUESTED"] \| None,
		occurred_at: str,
		) -> None:
		if self._callback_registration_repo is None:
		return
		registration = await self._callback_registration_repo.get_by_operation_id(operation_id)
		if registration is None or not registration.is_active:
		return
		if self._callback_delivery_port is None or self._callback_delivery_repo is None:
		raise RuntimeError("CallbackDeliveryPort/CallbackDeliveryRepository is not available")

		source = f"{get_settings().public_base_url}/qod/v0/sessions/{session_id}"
		cloud_event = build_qos_status_changed_event(
		session_id=session_id,
		qos_status=qos_status,
		status_info=status_info,
		source=source,
		occurred_at=occurred_at,
		)

		prior_attempts = await self._callback_delivery_repo.list_by_callback_registration_id(
		registration.id
		)
		last_error: Optional[str] = None
		try:
		await self._callback_delivery_port.deliver(registration.sink, cloud_event)
		state = "delivered"
		except Exception as exc:
		state = "failed"
		last_error = str(exc)
		logger.warning("qod_callback_delivery_failed", sink=registration.sink, error=str(exc))

		await self._callback_delivery_repo.save(
		CallbackDelivery(
		id=uuid4(),
		callback_registration_id=registration.id,
		operation_id=operation_id,
		attempt=len(prior_attempts) + 1,
		state=state,
		last_error=last_error,
		)
		)

		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")
		@@ -177,12 +260,59 @@ class QualityOnDemandService:

		instance = event.instances[0] if event.instances else None
		update: dict[str, object]
		qos_status: Literal["AVAILABLE", "UNAVAILABLE"]
		status_info: Literal["NETWORK_TERMINATED"] \| None
		if instance is not None and instance.status == "completed":
		update = {"state": QodSessionState.AVAILABLE, "external_ref": instance.external_ref}
		qos_status, status_info = "AVAILABLE", None
		else:
		update = {"state": QodSessionState.UNAVAILABLE}
		qos_status, status_info = "UNAVAILABLE", "NETWORK_TERMINATED"
		await self._qod_session_repo.save(qod_session.model_copy(update=update))

		await self._deliver_qos_status_changed(
		operation_id=operation_id,
		session_id=qod_session.session_id,
		qos_status=qos_status,
		status_info=status_info,
		occurred_at=event.completed_at,
		)

		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")

		operation_id = UUID(event.operation_id)
		operation = await self._operation_repo.get_by_id(operation_id)
		if operation is None or operation.operation_type != OperationType.NETWORK_CAPABILITY:
		return

		qod_session = await self._qod_session_repo.get_by_operation_id(operation_id)
		if qod_session is None:
		return

		qos_status = (event.metadata or {}).get("qos_status")
		if qos_status not in ("AVAILABLE", "UNAVAILABLE"):
		logger.warning(
		"operation_status_unrecognized_qos_status",
		operation_id=event.operation_id,
		metadata=event.metadata,
		)
		return

		status = (
		QodSessionState.AVAILABLE if qos_status == "AVAILABLE" else QodSessionState.UNAVAILABLE
		)
		await self._qod_session_repo.save(qod_session.model_copy(update={"state": status}))

		await self._deliver_qos_status_changed(
		operation_id=operation_id,
		session_id=qod_session.session_id,
		qos_status=qos_status,
		status_info="NETWORK_TERMINATED" if qos_status == "UNAVAILABLE" else None,
		occurred_at=event.emitted_at,
		)

		async def get_session(
		self,
		session_id: str,

tests/unit/conftest.py

+2 −5

Original line number	Diff line number	Diff line
		@@ -123,11 +123,6 @@ def live_srm(
		return fake_srm


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


		@pytest.fixture()
		def qod_session_repo() -> FakeQodSessionRepository:
		return FakeQodSessionRepository()
		@@ -169,12 +164,14 @@ def qod_service(
		fake_bus: FakeDataBus,
		operation_repo: FakeOperationRepository,
		qod_session_repo: FakeQodSessionRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		) -> QualityOnDemandService:
		return QualityOnDemandService(
		srm_client=fake_srm,
		publisher=fake_bus,
		operation_repo=operation_repo,
		qod_session_repo=qod_session_repo,
		callback_registration_repo=callback_registration_repo,
		)

Admin message