Commit de592200 authored by George Papathanail's avatar George Papathanail
Browse files

feat(qod): serve v1 API only, track session timing, migrate tests

parent 984da2eb
Loading
Loading
Loading
Loading
Loading
+11 −0
Changes for src/open_exposure_gateway/adapters/database/core.py: 11 added lines, 0 removed lines.
Original line number Diff line number Diff line
from typing import Literal

from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
@@ -23,6 +24,16 @@ async def build_engine_and_session_maker(
    return engine, session_maker


# create_all never alters existing tables; columns added after first release go here.
_ADDED_COLUMNS = (
    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS started_at TIMESTAMP WITH TIME ZONE",
    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS ended_at TIMESTAMP WITH TIME ZONE",
    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS status_info VARCHAR(32)",
)


async def schema_initialization(engine: AsyncEngine) -> None:
    async with engine.begin() as conn:
        await conn.run_sync(get_metadata().create_all)
        for statement in _ADDED_COLUMNS:
            await conn.execute(text(statement))
+8 −0
Changes for src/open_exposure_gateway/adapters/database/mappers.py: 8 added lines, 0 removed lines.
Original line number Diff line number Diff line
from typing import cast
from uuid import UUID

from open_exposure_gateway.adapters.database.sql import (
@@ -18,6 +19,7 @@ from open_exposure_gateway.domain.models import (
    Operation,
    QodSession,
)
from open_exposure_gateway.domain.models.qod_sessions.models import QodStatusInfo


class AppRegistrationMapper:
@@ -153,6 +155,9 @@ class QodSessionMapper:
            external_ref=row.external_ref,
            device_ports=row.device_ports,
            application_server_ports=row.application_server_ports,
            started_at=row.started_at,
            ended_at=row.ended_at,
            status_info=cast(QodStatusInfo | None, row.status_info),
            created_at=row.created_at,
            updated_at=row.updated_at,
        )
@@ -169,6 +174,9 @@ class QodSessionMapper:
            external_ref=domain.external_ref,
            device_ports=domain.device_ports,
            application_server_ports=domain.application_server_ports,
            started_at=domain.started_at,
            ended_at=domain.ended_at,
            status_info=domain.status_info,
        )


+3 −0
Changes for src/open_exposure_gateway/adapters/database/sql.py: 3 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -224,3 +224,6 @@ class QodSessionRow(AuditedMixin, Base):
    external_ref: Mapped[str | None] = mapped_column(String(255))
    device_ports: Mapped[dict[str, object] | None] = mapped_column(JSONB)
    application_server_ports: Mapped[dict[str, object] | None] = mapped_column(JSONB)
    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    status_info: Mapped[str | None] = mapped_column(String(32))
+0 −158
Changes for src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/router.py: 0 added lines, 158 removed lines.
Original line number Diff line number Diff line
from typing import Annotated, Any, Optional

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 (
    QualityOnDemandService,
)
from open_exposure_gateway.core.authorization import (
    require_role,
    require_scope,
)
from open_exposure_gateway.core.exceptions import (
    BadRequestException,
    ConflictException,
    DownstreamServiceException,
    ForbiddenException,
    NotFoundException,
    NotImplementedException,
    UnauthorizedException,
)
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).
BASE_PATH = "/qod/v0"

router = APIRouter(
    prefix=BASE_PATH,
    dependencies=[Depends(require_role), Depends(require_scope)],
)

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
# (it's the unhandled-exception catch-all in error_handlers.py).
_ERROR_RESPONSES: dict[int | str, dict[str, Any]] = {
    exc_cls().status_code: {"model": ErrorInfo, "description": exc_cls().message}
    for exc_cls in (
        BadRequestException,
        UnauthorizedException,
        ForbiddenException,
        NotFoundException,
        ConflictException,
        NotImplementedException,
        DownstreamServiceException,
    )
}
_ERROR_RESPONSES[500] = {"model": ErrorInfo, "description": "Internal server error"}


def _responses(*codes: int) -> dict[int | str, dict[str, Any]]:
    return {code: _ERROR_RESPONSES[code] for code in codes}


@router.post(
    "/sessions",
    tags=["QoS Sessions"],
    summary="Creates a new session",
    response_model=SessionInfo,
    response_model_exclude_none=True,
    status_code=status.HTTP_201_CREATED,
    responses=_responses(400, 401, 403, 409, 500, 501, 503),
)
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=caller.x_correlator,
    )


@router.get(
    "/sessions/{sessionId}",
    tags=["QoS Sessions"],
    summary="Get QoS session information",
    response_model=SessionInfo,
    response_model_exclude_none=True,
    responses=_responses(400, 401, 403, 404, 500, 503),
)
async def get_qod_session(
    sessionId: str,
    service: QoDService,
    caller: Caller,
    x_correlator: XCorrelatorHeader = None,
) -> Any:
    return await service.get_session(
        session_id=sessionId,
        x_correlator=caller.x_correlator,
    )


@router.delete(
    "/sessions/{sessionId}",
    tags=["QoS Sessions"],
    summary="Delete a QoS session",
    status_code=status.HTTP_204_NO_CONTENT,
    responses=_responses(400, 401, 403, 404, 500, 503),
)
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=caller.x_correlator,
    )
    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"],
    summary="Get All QoS Profiles",
    responses=_responses(401, 403, 404, 500, 501, 503),
)
async def get_qos_profiles(
    name: Annotated[Optional[str], Query()] = None,
    status: Annotated[Optional[str], Query()] = None,
) -> Any:
    raise NotImplementedException(message="QoS profile catalog is not implemented in this release")


@router.get(
    "/qos-profiles/{name}",
    tags=["QoS Profiles"],
    summary="Get QoS Profile for a given name",
    responses=_responses(400, 401, 403, 404, 500, 501, 503),
)
async def get_qos_profile(name: str) -> Any:
    raise NotImplementedException(message="QoS profile catalog is not implemented in this release")
+0 −122
Changes for src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/schemas.py: 0 added lines, 122 removed lines.
Original line number Diff line number Diff line
from enum import StrEnum
from typing import Literal, Optional
from uuid import UUID

from pydantic import BaseModel, Field, field_validator, model_validator

from open_exposure_gateway.api.camara.common import validated_ipv4, validated_ipv6

SOURCE_API = "quality-on-demand"
SOURCE_API_VERSION = "0.10.1"


class QosStatus(StrEnum):
    REQUESTED = "REQUESTED"
    AVAILABLE = "AVAILABLE"
    UNAVAILABLE = "UNAVAILABLE"


class DeviceIpv4Addr(BaseModel):
    publicAddress: Optional[str] = None
    privateAddress: Optional[str] = None
    publicPort: Optional[int] = Field(default=None, ge=0, le=65535)

    @field_validator("publicAddress", "privateAddress")
    @classmethod
    def _valid_ipv4(cls, value: Optional[str]) -> Optional[str]:
        return validated_ipv4(value)

    @model_validator(mode="after")
    def _require_public_plus_one(self) -> "DeviceIpv4Addr":
        if self.publicAddress is None:
            raise ValueError("publicAddress is required")
        if self.privateAddress is None and self.publicPort is None:
            raise ValueError("at least one of privateAddress or publicPort is required")
        return self


class Device(BaseModel):
    phoneNumber: Optional[str] = Field(default=None, pattern=r"^\+?[0-9]{5,15}$")
    networkAccessIdentifier: Optional[str] = None
    ipv4Address: Optional[DeviceIpv4Addr] = None
    ipv6Address: Optional[str] = None

    @field_validator("ipv6Address")
    @classmethod
    def _valid_ipv6(cls, value: Optional[str]) -> Optional[str]:
        return validated_ipv6(value)

    @model_validator(mode="after")
    def _require_at_least_one_identifier(self) -> "Device":
        if not any(
            (self.phoneNumber, self.networkAccessIdentifier, self.ipv4Address, self.ipv6Address)
        ):
            raise ValueError("at least one device identifier must be provided")
        return self


class ApplicationServer(BaseModel):
    ipv4Address: Optional[str] = None
    ipv6Address: Optional[str] = None

    @model_validator(mode="after")
    def _require_at_least_one_address(self) -> "ApplicationServer":
        if not any((self.ipv4Address, self.ipv6Address)):
            raise ValueError("at least one of ipv4Address or ipv6Address must be provided")
        return self


class PortRange(BaseModel):
    from_: int = Field(alias="from", ge=0, le=65535)
    to: int = Field(ge=0, le=65535)

    model_config = {
        "populate_by_name": True,
    }


class PortsSpec(BaseModel):
    ranges: Optional[list[PortRange]] = None
    ports: Optional[list[int]] = None

    @model_validator(mode="after")
    def _require_ranges_or_ports(self) -> "PortsSpec":
        if not self.ranges and not self.ports:
            raise ValueError("at least one of ranges or ports must be provided")
        return self


class Webhook(BaseModel):
    notificationUrl: str
    notificationAuthToken: Optional[str] = Field(default=None, min_length=20, max_length=256)


class Message(BaseModel):
    severity: Literal["INFO", "WARNING"]
    description: str


class BaseSessionInfo(BaseModel):
    device: Device
    applicationServer: ApplicationServer
    devicePorts: Optional[PortsSpec] = None
    applicationServerPorts: Optional[PortsSpec] = None
    qosProfile: str = Field(min_length=3, max_length=256, pattern=r"^[a-zA-Z0-9_.-]+$")
    webhook: Optional[Webhook] = None


class CreateSession(BaseSessionInfo):
    duration: int = Field(default=86400, ge=1, le=86400)


class SessionInfo(BaseSessionInfo):
    sessionId: UUID
    duration: int = Field(ge=1)
    startedAt: int
    expiresAt: int
    qosStatus: QosStatus
    messages: Optional[list[Message]] = None


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