feat: add EAM deployment stub endpoints and QoD session-extend endpoint with request validation (b7f9a492) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/api/camara/edge_application_management/vwip/router.py

+90 −0

Original line number	Diff line number	Diff line
		@@ -281,3 +281,93 @@ async def delete_app_instance(
		x_correlator=caller.x_correlator,
		)
		return Response(status_code=status.HTTP_202_ACCEPTED)


		@router.post(
		"/deployments",
		tags=["Application"],
		summary="Deploy an Application",
		responses=_responses(
		BadRequestException(),
		UnauthorizedException(),
		ForbiddenException(),
		AlreadyExistsException(),
		NotImplementedException(),
		DownstreamServiceException(),
		),
		)
		async def create_app_deployment() -> Any:
		raise NotImplementedException(message="Multi-zone deployment is not implemented")


		@router.get(
		"/deployments",
		tags=["Application"],
		summary="Retrieve a list of Application Deployment for a given App",
		responses=_responses(
		BadRequestException(),
		UnauthorizedException(),
		ForbiddenException(),
		NotImplementedException(),
		DownstreamServiceException(),
		),
		)
		async def get_app_deployments(
		appId: Annotated[Optional[UUID], Query()] = None,
		appDeploymentId: Annotated[Optional[UUID], Query()] = None,
		) -> Any:
		raise NotImplementedException(message="Multi-zone deployment is not implemented")


		@router.delete(
		"/deployments/{appDeploymentId}",
		tags=["Application"],
		summary="Terminate an Application Deployment",
		responses=_responses(
		BadRequestException(),
		UnauthorizedException(),
		ForbiddenException(),
		NotFoundException(),
		NotImplementedException(),
		DownstreamServiceException(),
		),
		)
		async def delete_app_deployment(appDeploymentId: UUID) -> Any:
		raise NotImplementedException(message="Multi-zone deployment is not implemented")


		@router.patch(
		"/deployments/{appDeploymentId}",
		tags=["Application"],
		summary="Update an Application Deployment",
		responses=_responses(
		BadRequestException(),
		UnauthorizedException(),
		ForbiddenException(),
		NotFoundException(),
		NotImplementedException(),
		DownstreamServiceException(),
		),
		)
		async def update_app_deployment(appDeploymentId: UUID) -> Any:
		raise NotImplementedException(message="Multi-zone deployment is not implemented")


		@router.get(
		"/clusters",
		tags=["Cluster"],
		summary="Retrieve a list of the available clusters filtered by the optional query parameters",
		responses=_responses(
		BadRequestException(),
		UnauthorizedException(),
		ForbiddenException(),
		NotImplementedException(),
		DownstreamServiceException(),
		),
		)
		async def get_clusters(
		region: Annotated[Optional[str], Query(max_length=64)] = None,
		clusterRef: Annotated[Optional[UUID], Query()] = None,
		edgeCloudZoneId: Annotated[Optional[UUID], Query()] = None,
		) -> Any:
		raise NotImplementedException(message="Cluster catalog is not implemented")

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

+11 −0

Original line number	Diff line number	Diff line
		@@ -5,6 +5,7 @@ from fastapi import APIRouter, Depends, Query, Response, status
		from open_exposure_gateway.api.camara.common import XCorrelatorHeader
		from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (
		CreateSession,
		ExtendSessionDuration,
		SessionInfo,
		)
		from open_exposure_gateway.application.services.quality_on_demand_service import (
		@@ -116,6 +117,16 @@ async def delete_qod_session(
		return Response(status_code=status.HTTP_204_NO_CONTENT)


		@router.post(
		"/sessions/{sessionId}/extend",
		tags=["QoS Sessions"],
		summary="Extend the duration of an active session",
		responses=_responses(400, 401, 403, 404, 500, 501, 503),
		)
		async def extend_qod_session_duration(sessionId: str, request: ExtendSessionDuration) -> Any:
		raise NotImplementedException(message="Session duration extension is not implemented")


		@router.get(
		"/qos-profiles",
		tags=["QoS Profiles"],

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

+4 −0

Original line number	Diff line number	Diff line
		@@ -101,3 +101,7 @@ class SessionInfo(BaseSessionInfo):
		expiresAt: int
		qosStatus: QosStatus
		messages: Optional[list[Message]] = None


		class ExtendSessionDuration(BaseModel):
		requestedAdditionalDuration: int = Field(ge=1, le=86399)

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+26 −4

Original line number	Diff line number	Diff line
		@@ -20,7 +20,11 @@ from open_exposure_gateway.core.config import (
		DEFAULT_QOD_SERVICE_SPECIFICATION_ID,
		get_settings,
		)
		from open_exposure_gateway.core.exceptions import DownstreamServiceException, NotFoundException
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		DownstreamServiceException,
		NotFoundException,
		)
		from open_exposure_gateway.domain.edge_application_management import SRMOperationCompleted
		from open_exposure_gateway.domain.models import (
		CallbackDelivery,
		@@ -50,6 +54,11 @@ _OPERATION_COMPLETION_STATUS_MAP: dict[str, OperationStatus] = {
		"failed": OperationStatus.FAILED,
		}

		_QOS_STATUS_INFO_MAP: dict[str, Literal["DURATION_EXPIRED", "NETWORK_TERMINATED"]] = {
		"duration_expired": "DURATION_EXPIRED",
		"network_terminated": "NETWORK_TERMINATED",
		}

		_TERMINAL_STATES = frozenset(
		{QodSessionState.DELETION_REQUESTED, QodSessionState.DELETED, QodSessionState.ERROR}
		)
		@@ -76,6 +85,12 @@ class QualityOnDemandService:
		self._callback_delivery_port = callback_delivery_port
		self._service_specification_id = service_specification_id

		def _parse_session_id(self, session_id: str) -> UUID:
		try:
		return UUID(session_id)
		except ValueError as exc:
		raise BadRequestException(message=f"Malformed session ID: {session_id}") from exc

		def _new_operation_metadata(self, x_correlator: Optional[str]) -> tuple[UUID, str, str]:
		operation_id = uuid4()
		correlation_id = x_correlator or str(uuid4())
		@@ -362,11 +377,18 @@ class QualityOnDemandService:
		if prior_state == status:
		return

		status_info: Literal["DURATION_EXPIRED", "NETWORK_TERMINATED"] \| None = None
		if qos_status == "UNAVAILABLE":
		reason = (event.metadata or {}).get("qos_status_info")
		status_info = _QOS_STATUS_INFO_MAP.get(
		reason if isinstance(reason, str) else "", "NETWORK_TERMINATED"
		)

		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,
		status_info=status_info,
		occurred_at=event.emitted_at,
		)

		@@ -378,7 +400,7 @@ class QualityOnDemandService:
		if self._qod_session_repo is None:
		raise RuntimeError("QodSession repository is not available")

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

		@@ -398,7 +420,7 @@ class QualityOnDemandService:
		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))
		qod_session = await self._qod_session_repo.get_by_id(self._parse_session_id(session_id))
		if qod_session is None or qod_session.state in _TERMINAL_STATES:
		raise NotFoundException(message=f"Session {session_id} not found")

tests/conformance/conftest.py

+27 −1

Original line number	Diff line number	Diff line
		@@ -30,11 +30,15 @@ from tests.conformance.harness import app
		from tests.unit.fakes import (
		FakeAppInstanceRepository,
		FakeAppRegistrationRepository,
		FakeCallbackDeliveryRepository,
		FakeCallbackRegistrationRepository,
		FakeDataBus,
		FakeOperationRepository,
		FakeQodCallbackDeliveryPort,
		FakeQodSessionRepository,
		FakeSRMClient,
		wire_operation_consumer,
		wire_qod_operation_consumer,
		wire_srm_worker,
		)

		@@ -66,7 +70,29 @@ def service_overrides() -> Generator[None, None, None]:
		app_instance_repo=app_instance_repo,
		callback_registration_repo=FakeCallbackRegistrationRepository(),
		)
		app.dependency_overrides[get_qod_service] = lambda: QualityOnDemandService(srm_client=srm)

		qod_operation_repo = FakeOperationRepository()
		qod_session_repo = FakeQodSessionRepository()
		qod_callback_registration_repo = FakeCallbackRegistrationRepository()
		qod_callback_delivery_repo = FakeCallbackDeliveryRepository()
		qod_callback_delivery_port = FakeQodCallbackDeliveryPort()
		wire_qod_operation_consumer(
		bus,
		qod_operation_repo,
		qod_session_repo,
		qod_callback_registration_repo,
		qod_callback_delivery_repo,
		qod_callback_delivery_port,
		)
		app.dependency_overrides[get_qod_service] = lambda: QualityOnDemandService(
		srm_client=srm,
		publisher=bus,
		operation_repo=qod_operation_repo,
		qod_session_repo=qod_session_repo,
		callback_registration_repo=qod_callback_registration_repo,
		callback_delivery_repo=qod_callback_delivery_repo,
		callback_delivery_port=qod_callback_delivery_port,
		)
		app.dependency_overrides[get_publisher] = lambda: bus
		yield
		app.dependency_overrides.clear()