Commit 31e28fd6 authored by George Papathanail's avatar George Papathanail
Browse files

feat: make POST /sessions async via the databus, tests added

parent d59f7812

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

+5 −1
Original line number Diff line number Diff line
@@ -19,7 +19,7 @@ from open_exposure_gateway.core.exceptions import ( 
    NotImplementedException,

    UnauthorizedException,

)

from open_exposure_gateway.dependencies import get_qod_service

from open_exposure_gateway.dependencies import CallerContext, get_caller_context, get_qod_service

from open_exposure_gateway.schemas.common import ErrorInfo



# CAMARA base path: the spec serves at {apiRoot}/qod/v0 (wire version of v0.10.1).
@@ -28,6 +28,7 @@ BASE_PATH = "/qod/v0" 
router = APIRouter(prefix=BASE_PATH)



QoDService = Annotated[QualityOnDemandService, Depends(get_qod_service)]

Caller = Annotated[CallerContext, Depends(get_caller_context)]



# OpenAPI response docs derived from core.exceptions' status_code/message

# defaults, so codes aren't re-listed. 500 has no dedicated exception class
@@ -63,10 +64,13 @@ def _responses(*codes: int) -> dict[int | str, dict[str, Any]]: 
async def create_qod_session(

    request: CreateSession,

    service: QoDService,

    caller: Caller,

    x_correlator: XCorrelatorHeader = None,

) -> Any:

    return await service.create_session(

        request=request,

        tenant_id=caller.tenant_id,

        app_provider_id=caller.app_provider_id,

        x_correlator=x_correlator,

    )

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+107 −4
Original line number Diff line number Diff line 
from datetime import datetime, timezone

from typing import Optional

from uuid import UUID, uuid4



from pydantic import BaseModel



from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (

    CreateSession,

    QosStatus,

    SessionInfo,

)

from open_exposure_gateway.application.mappers.quality_on_demand_mapper import (

    build_activate_command,

)

from open_exposure_gateway.core.exceptions import DownstreamServiceException

from open_exposure_gateway.domain.models import (

    Operation,

    OperationStatus,

    OperationType,

    QodSession,

    QodSessionState,

)

from open_exposure_gateway.domain.quality_on_demand import Subject

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.srm_port import SRMClientPort





class QualityOnDemandService:

    def __init__(self, srm_client: SRMClientPort) -> None:

    def __init__(

        self,

        srm_client: SRMClientPort,

        publisher: Optional[DataBusPort] = None,

        operation_repo: Optional[OperationRepository] = None,

        qod_session_repo: Optional[QodSessionRepository] = None,

    ) -> None:

        self.srm_client = srm_client

        self._publisher = publisher

        self._operation_repo = operation_repo

        self._qod_session_repo = qod_session_repo



    def _new_operation_metadata(self, x_correlator: Optional[str]) -> tuple[UUID, str, str]:

        operation_id = uuid4()

        correlation_id = x_correlator or str(uuid4())

        requested_at = datetime.now(timezone.utc).isoformat()

        return operation_id, correlation_id, requested_at



    async def _publish(self, subject: Subject, command: BaseModel, error_msg: str) -> None:

        if self._publisher is None:

            raise RuntimeError("DataBus publisher is not available")

        try:

            await self._publisher.publish(subject, command.model_dump(mode="json"))

        except Exception as exc:

            raise DownstreamServiceException(error_msg) from exc



    async def create_session(

        self,

        request: CreateSession,

        tenant_id: str,

        app_provider_id: str,

        x_correlator: Optional[str] = None,

    ) -> SessionInfo:

        return await self.srm_client.create_qod_session(

            payload=request.model_dump(mode="json"),

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



        operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)

        session_id = uuid4()

        service_specification_id = uuid4()



        command = build_activate_command(

            request=request,

            operation_id=operation_id,

            correlation_id=correlation_id,

            requested_at=requested_at,

            service_specification_id=service_specification_id,

            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,

                status=OperationStatus.PENDING,

                subject=Subject.TASK_ACTIVATE,

                metadata={

                    "qos_profile": request.qosProfile,

                    "service_specification_id": str(service_specification_id),

                },

            )

        )

        await self._qod_session_repo.save(

            QodSession(

                session_id=session_id,

                operation_id=operation_id,

                service_specification_id=service_specification_id,

                qos_profile=request.qosProfile,

                duration_seconds=request.duration,

                state=QodSessionState.REQUESTED,

            )

        )



        await self._publish(

            Subject.TASK_ACTIVATE,

            command,

            "Failed to publish QoD session activation",

        )



        started_at = int(datetime.now(timezone.utc).timestamp())

        return SessionInfo(

            sessionId=session_id,

            device=request.device,

            applicationServer=request.applicationServer,

            devicePorts=request.devicePorts,

            applicationServerPorts=request.applicationServerPorts,

            qosProfile=request.qosProfile,

            webhook=request.webhook,

            duration=request.duration,

            startedAt=started_at,

            expiresAt=started_at + request.duration,

            qosStatus=QosStatus.REQUESTED,

        )



    async def get_session(

src/open_exposure_gateway/dependencies.py

+12 −1
Original line number Diff line number Diff line
@@ -18,6 +18,9 @@ from open_exposure_gateway.adapters.database.repos.callback_registrations import 
from open_exposure_gateway.adapters.database.repos.operations import (

    SqlOperationRepository,

)

from open_exposure_gateway.adapters.database.repos.qod_sessions import (

    SqlQodSessionRepository,

)

from open_exposure_gateway.api.camara.common import XCorrelatorHeader

from open_exposure_gateway.application.services.edge_application_management_service import (

    EdgeApplicationManagementService,
@@ -29,6 +32,7 @@ from open_exposure_gateway.core.state import AppState 
from open_exposure_gateway.ports.database.callbacks import CallbackRegistrationRepository

from open_exposure_gateway.ports.database.instances import AppInstanceRepository

from open_exposure_gateway.ports.database.operations import OperationRepository

from open_exposure_gateway.ports.database.qod_sessions import QodSessionRepository

from open_exposure_gateway.ports.database.registration import AppRegistrationRepository

from open_exposure_gateway.ports.databus_port import DataBusPort

from open_exposure_gateway.ports.srm_port import SRMClientPort
@@ -111,6 +115,10 @@ def get_callback_registration_repo(session: SessionDep) -> CallbackRegistrationR 
    return SqlCallbackRegistrationRepository(session)





def get_qod_session_repo(session: SessionDep) -> QodSessionRepository:

    return SqlQodSessionRepository(session)





def get_edge_app_service(

    srm: SRMClientPort = Depends(get_client),

    publisher: DataBusPort = Depends(get_publisher),
@@ -133,5 +141,8 @@ def get_edge_app_service( 


def get_qod_service(

    srm: SRMClientPort = Depends(get_client),

    publisher: DataBusPort = Depends(get_publisher),

    operation_repo: OperationRepository = Depends(get_operation_repo),

    qod_session_repo: QodSessionRepository = Depends(get_qod_session_repo),

) -> QualityOnDemandService:

    return QualityOnDemandService(srm)
 
    return QualityOnDemandService(srm, publisher, operation_repo, qod_session_repo)

tests/unit/conftest.py

+23 −2
Original line number Diff line number Diff line
@@ -29,6 +29,7 @@ from tests.unit.fakes import ( 
    FakeCallbackRegistrationRepository,

    FakeDataBus,

    FakeOperationRepository,

    FakeQodSessionRepository,

    FakeSRMClient,

    wire_operation_consumer,

    wire_srm_worker,
@@ -121,6 +122,16 @@ def live_srm( 
    return fake_srm





@pytest.fixture()

def operation_repo() -> FakeOperationRepository:

    return FakeOperationRepository()





@pytest.fixture()

def qod_session_repo() -> FakeQodSessionRepository:

    return FakeQodSessionRepository()





@pytest.fixture()

def eam_service(

    fake_srm: FakeSRMClient,
@@ -141,8 +152,18 @@ def eam_service( 




@pytest.fixture()

def qod_service(fake_srm: FakeSRMClient) -> QualityOnDemandService:

    return QualityOnDemandService(srm_client=fake_srm)

def qod_service(

    fake_srm: FakeSRMClient,

    fake_bus: FakeDataBus,

    operation_repo: FakeOperationRepository,

    qod_session_repo: FakeQodSessionRepository,

) -> QualityOnDemandService:

    return QualityOnDemandService(

        srm_client=fake_srm,

        publisher=fake_bus,

        operation_repo=operation_repo,

        qod_session_repo=qod_session_repo,

    )





@pytest.fixture()

tests/unit/fakes.py

+16 −0
Original line number Diff line number Diff line
@@ -51,6 +51,7 @@ from open_exposure_gateway.domain.models import ( 
    CallbackDelivery,

    CallbackRegistration,

    Operation,

    QodSession,

)

from open_exposure_gateway.ports.callback_delivery_port import CallbackDeliveryPort

from open_exposure_gateway.ports.database.callbacks import (
@@ -59,6 +60,7 @@ from open_exposure_gateway.ports.database.callbacks import ( 
)

from open_exposure_gateway.ports.database.instances import AppInstanceRepository

from open_exposure_gateway.ports.database.operations import OperationRepository

from open_exposure_gateway.ports.database.qod_sessions import QodSessionRepository

from open_exposure_gateway.ports.database.registration import AppRegistrationRepository



Handler = Callable[[dict[str, Any]], Awaitable[None]]
@@ -388,6 +390,20 @@ class FakeAppInstanceRepository(AppInstanceRepository): 
        return stored.model_copy(deep=True)





class FakeQodSessionRepository(QodSessionRepository):

    def __init__(self) -> None:

        self.rows: dict[UUID, QodSession] = {}



    async def get_by_id(self, session_id: UUID) -> QodSession | None:

        found = self.rows.get(session_id)

        return found.model_copy(deep=True) if found is not None else None



    async def save(self, qod_session: QodSession) -> QodSession:

        stored = qod_session.model_copy(deep=True)

        self.rows[stored.session_id] = stored

        return stored.model_copy(deep=True)





class FakeCallbackRegistrationRepository(CallbackRegistrationRepository):

    def __init__(self) -> None:

        self.rows: dict[UUID, CallbackRegistration] = {}