Feat/qod v1

George Papathanailrequested to merge
feat/qod-v1 into develop

QoD: move to CAMARA Quality-On-Demand v1.1.0

Replaces the QoD v0.10.1 API with CAMARA Quality-On-Demand v1.1.0, served at /quality-on-demand/v1. v0 (/qod/v0) is removed, not served side by side.

Breaking changes

  • New base path: /quality-on-demand/v1 (was /qod/v0).
  • New request fields:
    • webhook.notificationUrl / notificationAuthToken → sink / sinkCredential. sink must be https://.
    • duration is now required.
  • New scopes, one per operation as the spec declares:
    • quality-on-demand:sessions:create, :read, :delete, :update, :retrieve-by-device.
    • These replace qod-sessions-read|write|delete.
    • Client-credentials callers no longer skip the scope check. Tokens must carry the new scope names, or callers get 403 when enforcement is on.
  • Removed: the v0 qos-profiles routes, which are not part of the v1 API.

API

  • Schemas and router vendored from the v1.1.0 spec (api/camara/quality_on_demand/v1_1_0/).
  • New stub: POST /retrieve-sessions returns 501, like the existing /extend stub.
  • Error codes: added the v1 codes INVALID_SINK, INVALID_CREDENTIAL, INVALID_TOKEN, QUALITY_ON_DEMAND.DURATION_OUT_OF_RANGE, QUALITY_ON_DEMAND.QOS_PROFILE_NOT_APPLICABLE and QUALITY_ON_DEMAND.SESSION_EXTENSION_NOT_ALLOWED. BadRequestException and ConflictException now take an error_code.

Session behaviour

  • Session timing: qod_sessions gains started_at, ended_at and status_info. They are set when a session becomes AVAILABLE or UNAVAILABLE, and GET fills in startedAt, expiresAt, duration and statusInfo as the spec describes for each status.
  • Ownership: GET and DELETE /sessions/{id} return 404 unless the caller's app_provider_id owns the session (checked through its activate operation).
  • GET returns sink when one was registered at create time.
  • Device identifiers:
    • A networkAccessIdentifier sent alongside a supported identifier is ignored, not sent to SRM.
    • 422 UNSUPPORTED_IDENTIFIER is returned only when it is the only identifier.
    • A missing device gives 422 MISSING_IDENTIFIER.
  • SRM activate commands now carry source_spec.version = 1.1.0.

Callbacks

  • The CloudEvent type is now org.camaraproject.quality-on-demand.v1.qos-status-changed.
  • Empty fields are left out of the payload.
  • Each callback carries an x-correlator header.

Database

create_all does not alter existing tables, so schema_initialization now runs ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS … for the three new columns. Existing databases upgrade in place on startup.

Known gaps (not in this MR)

  • sinkCredential returns 501, because callback credentials can't be stored yet. The spec doesn't document 501 on POST /sessions, so test_qod_conformance[POST /sessions] fails. /extend and /retrieve-sessions fail that test for the same reason.
  • DURATION_EXPIRED is never triggered.
  • No 409 check for an existing session on the same device.
  • No QoS profile or duration validation at create time.
  • A sink answering 410 is ignored.
  • No support for three-legged tokens.

Testing

  • Unit tests were moved to v1 and extended to cover ownership, sink on GET, identifier handling, session timing and the callback format.
  • pytest tests/unit: 899 passed, 1 skipped.
Edited by George Papathanail

Merge request reports

src/open_exposure_gateway/adapters/database/repos/qod_sessions.py

+68 −3
Changes for src/open_exposure_gateway/adapters/database/repos/qod_sessions.py: 68 added lines, 3 removed lines.
Original line number Diff line number Diff line 
from datetime import datetime

from typing import Any

from uuid import UUID



from sqlalchemy import select

from sqlalchemy import and_, literal_column, or_, select, type_coerce

from sqlalchemy.dialects.postgresql import JSONB

from sqlalchemy.ext.asyncio import AsyncSession



from open_exposure_gateway.adapters.database.mappers import QodSessionMapper

from open_exposure_gateway.adapters.database.sql import QodSessionRow

from open_exposure_gateway.domain.models import QodSession

from open_exposure_gateway.adapters.database.sql import OperationRow, QodSessionRow

from open_exposure_gateway.domain.models import QodSession, QodSessionState

from open_exposure_gateway.ports.database.qod_sessions import QodSessionRepository



_ACTIVE_STATES = (QodSessionState.REQUESTED, QodSessionState.AVAILABLE)

_DEVICE_IDENTIFIERS = ("phoneNumber", "ipv4Address", "ipv6Address")





class SqlQodSessionRepository(QodSessionRepository):

    def __init__(self, session: AsyncSession) -> None:
@@ -23,6 +29,65 @@ class SqlQodSessionRepository(QodSessionRepository): 
        row = await self._session.scalar(stmt)

        return QodSessionMapper.to_domain(row) if row is not None else None



    async def find_active_by_device(

        self, device: dict[str, Any], app_provider_id: str

    ) -> QodSession | None:

        identifier_matches = [

            QodSessionRow.device[key] == type_coerce(device[key], JSONB)

            for key in _DEVICE_IDENTIFIERS

            if device.get(key) is not None

        ]

        if not identifier_matches:

            return None

        stmt = (

            select(QodSessionRow)

            .join(OperationRow, OperationRow.operation_id == QodSessionRow.operation_id)

            .where(

                OperationRow.app_provider_id == app_provider_id,

                or_(

                    QodSessionRow.state.in_(_ACTIVE_STATES),

                    and_(

                        QodSessionRow.state == QodSessionState.UNAVAILABLE,

                        QodSessionRow.status_info == "NETWORK_TERMINATED",

                    ),

                ),

                or_(*identifier_matches),

            )

            .limit(1)

        )

        row = await self._session.scalar(stmt)

        return QodSessionMapper.to_domain(row) if row is not None else None



    async def list_available_due(self, now: datetime) -> list[QodSession]:

        ends_at = QodSessionRow.started_at + QodSessionRow.duration_seconds * literal_column(

            "INTERVAL '1 second'"

        )

        stmt = (

            select(QodSessionRow)

            .where(

                QodSessionRow.state == QodSessionState.AVAILABLE,

                QodSessionRow.started_at.is_not(None),

                ends_at <= now,

            )

            .order_by(QodSessionRow.started_at)

        )

        rows = await self._session.scalars(stmt)

        return [QodSessionMapper.to_domain(row) for row in rows]



    async def list_network_terminated(self, ended_before: datetime) -> list[QodSession]:

        stmt = (

            select(QodSessionRow)

            .where(

                QodSessionRow.state == QodSessionState.UNAVAILABLE,

                QodSessionRow.status_info == "NETWORK_TERMINATED",

                QodSessionRow.ended_at.is_not(None),

                QodSessionRow.ended_at <= ended_before,

            )

            .order_by(QodSessionRow.ended_at)

        )

        rows = await self._session.scalars(stmt)

        return [QodSessionMapper.to_domain(row) for row in rows]



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

        merged = await self._session.merge(QodSessionMapper.to_row(qod_session))

        await self._session.flush()

src/open_exposure_gateway/adapters/database/core.py

+13 −0
Changes for src/open_exposure_gateway/adapters/database/core.py: 13 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,18 @@ 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)",

    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS device JSONB",

    "ALTER TABLE qod_sessions ADD COLUMN IF NOT EXISTS application_server JSONB",

)





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

src/open_exposure_gateway/adapters/database/mappers.py

+12 −0
Changes for src/open_exposure_gateway/adapters/database/mappers.py: 12 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 (
@@ -22,6 +23,7 @@ from open_exposure_gateway.domain.models import ( 
    QodSession,

    TrafficInfluence,

)

from open_exposure_gateway.domain.models.qod_sessions.models import QodStatusInfo





class AppRegistrationMapper:
@@ -157,6 +159,11 @@ class QodSessionMapper: 
            external_ref=row.external_ref,

            device_ports=row.device_ports,

            application_server_ports=row.application_server_ports,

            device=row.device,

            application_server=row.application_server,

            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,

        )
@@ -173,6 +180,11 @@ class QodSessionMapper: 
            external_ref=domain.external_ref,

            device_ports=domain.device_ports,

            application_server_ports=domain.application_server_ports,

            device=domain.device,

            application_server=domain.application_server,

            started_at=domain.started_at,

            ended_at=domain.ended_at,

            status_info=domain.status_info,

        )

src/open_exposure_gateway/adapters/database/sql.py

+5 −0
Changes for src/open_exposure_gateway/adapters/database/sql.py: 5 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -244,6 +244,11 @@ 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)

    device: Mapped[dict[str, object] | None] = mapped_column(JSONB)

    application_server: 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))





class TrafficInfluenceRow(AuditedMixin, Base):

src/open_exposure_gateway/adapters/http/qod_callback_client.py

+12 −3
Changes for src/open_exposure_gateway/adapters/http/qod_callback_client.py: 12 added lines, 3 removed lines.
Original line number Diff line number Diff line 
import httpx



from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.domain.errors import CallbackSinkGoneError

from open_exposure_gateway.domain.quality_on_demand import QosStatusChangedCloudEvent
@@ -10,11 +11,19 @@ class HttpQodCallbackClient: 
        self.timeout = settings.callback_settings.timeout



    async def deliver(

        self, sink: str, event: QosStatusChangedCloudEvent, bearer_token: str | None = None

        self,

        sink: str,

        event: QosStatusChangedCloudEvent,

        x_correlator: str,

        bearer_token: str | None = None,

    ) -> None:

        headers = {"Content-Type": "application/cloudevents+json"}

        headers = {"Content-Type": "application/cloudevents+json", "x-correlator": x_correlator}

        if bearer_token is not None:

            headers["Authorization"] = f"Bearer {bearer_token}"

        async with httpx.AsyncClient(timeout=self.timeout) as client:

            response = await client.post(sink, content=event.model_dump_json(), headers=headers)

            response = await client.post(

                sink, content=event.model_dump_json(exclude_none=True), headers=headers

            )

        if response.status_code == httpx.codes.GONE:

            raise CallbackSinkGoneError(f"sink answered {response.status_code}")

        response.raise_for_status()