Commit 7434f8ae authored by George Papathanail's avatar George Papathanail
Browse files

Merge branch 'feat/ti-devices' into 'develop' 

feat: implement POST /traffic-influence-devices

See merge request !52
parents 68ff79a6 c7f0f1be

src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/router.py

+25 −4
Changes for src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/router.py: 25 added lines, 4 removed lines.
Original line number Diff line number Diff line
@@ -61,6 +61,7 @@ _ERROR_RESPONSES: dict[int | str, dict[str, Any]] = { 
        DownstreamServiceException,

    )

}

_ERROR_RESPONSES[422] = {"model": ErrorInfo, "description": "Unprocessable content"}

_ERROR_RESPONSES[500] = {"model": ErrorInfo, "description": "Internal server error"}
@@ -170,12 +171,32 @@ async def delete_traffic_influence( 
    tags=["Traffic Influence API write"],

    summary="Creates a new TrafficInfluence resource influencing the traffic "

    "toward local instances of the Application for a specific user.",

    responses=_responses(400, 401, 403, 404, 409, 500, 501, 503),

    response_model=TrafficInfluence,

    response_model_exclude_none=True,

    status_code=status.HTTP_201_CREATED,

    responses=_responses(400, 401, 403, 404, 409, 422, 500, 503),

)

async def post_traffic_influence_device(request: PostTrafficInfluenceDevice) -> Any:

    raise NotImplementedException(

        message="Device-scoped traffic influence is not implemented in this release"

async def post_traffic_influence_device(

    request: PostTrafficInfluenceDevice,

    service: TrafficInfluenceServiceDep,

    caller: Caller,

    http_request: Request,

    response: Response,

    x_correlator: XCorrelatorHeader = None,

    idempotency_key: IdempotencyKeyHeader = None,

) -> Any:

    created = await service.create_traffic_influence_device(

        request=request,

        tenant_id=caller.tenant_id,

        app_provider_id=caller.app_provider_id,

        x_correlator=x_correlator,

        idempotency_key=idempotency_key,

    )

    base = str(http_request.base_url).rstrip("/")

    response.headers["Location"] = (

        f"{base}{BASE_PATH}/traffic-influences/{created.trafficInfluenceID}"

    )

    return created





@router.patch(

src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/schemas.py

+57 −10
Changes for src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/schemas.py: 57 added lines, 10 removed lines.
Original line number Diff line number Diff line
@@ -3,7 +3,9 @@ from enum import StrEnum 
from typing import Any, Literal, Optional

from uuid import UUID



from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, field_validator

from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, field_validator, model_validator



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



SOURCE_API = "traffic-influence"

SOURCE_API_VERSION = "0.10.0"
@@ -85,19 +87,64 @@ class PostTrafficInfluence(BaseTrafficInfluence): 
    pass





class Device(BaseModel):

    """CAMARA `Device` -- the subscriber a device-scoped policy applies to. Modelled

    loosely: `POST /traffic-influence-devices` is a 501 stub, so nothing reads these

    fields yet. Tighten to the full DeviceIpv4Addr/PhoneNumber shapes when it is served.

    """

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)



    phoneNumber: Optional[str] = None

    @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"^\+[1-9][0-9]{4,14}$")

    networkAccessIdentifier: Optional[str] = None

    ipv4Address: Optional[dict[str, Any]] = None

    ipv4Address: Optional[DeviceIpv4Addr] = None

    ipv6Address: Optional[str] = None





class PostTrafficInfluenceDevice(BaseTrafficInfluence):

    @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 DeviceResponse(Device):

    @model_validator(mode="after")

    def _require_exactly_one_identifier(self) -> "DeviceResponse":

        provided = [

            f

            for f in (

                self.phoneNumber,

                self.networkAccessIdentifier,

                self.ipv4Address,

                self.ipv6Address,

            )

            if f

        ]

        if len(provided) != 1:

            raise ValueError("exactly one device identifier must be provided")

        return self





class PostTrafficInfluenceDevice(PostTrafficInfluence):

    device: Optional[Device] = None

src/open_exposure_gateway/application/mappers/traffic_influence_mapper.py

+33 −1
Changes for src/open_exposure_gateway/application/mappers/traffic_influence_mapper.py: 33 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -6,13 +6,18 @@ from pydantic import ValidationError 
from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (

    SOURCE_API,

    SOURCE_API_VERSION,

    Device,

    DeviceResponse,

    PostTrafficInfluence,

)

from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (

    TrafficInfluence as TrafficInfluenceResponse,

)

from open_exposure_gateway.domain.models import TrafficInfluence as TrafficInfluenceRecord

from open_exposure_gateway.domain.quality_on_demand import SourceSpec

from open_exposure_gateway.domain.quality_on_demand import (

    NetworkCapabilityTargetDevice,

    SourceSpec,

)

from open_exposure_gateway.domain.traffic_influence import (

    NetworkCapabilityDeactivateTarget,

    NetworkCapabilityParameters,
@@ -28,6 +33,17 @@ from open_exposure_gateway.domain.traffic_influence import ( 
)





def build_device_response(device: Device) -> DeviceResponse | None:

    """CAMARA returns exactly one identifier: phone number, then IPv4, then IPv6."""

    if device.phoneNumber is not None:

        return DeviceResponse.model_construct(phoneNumber=device.phoneNumber)

    if device.ipv4Address is not None:

        return DeviceResponse.model_construct(ipv4Address=device.ipv4Address)

    if device.ipv6Address is not None:

        return DeviceResponse.model_construct(ipv6Address=device.ipv6Address)

    return None





def build_activate_command(

    request: PostTrafficInfluence,

    operation_id: UUID,
@@ -37,6 +53,7 @@ def build_activate_command( 
    service_instance_id: UUID,

    app_provider_id: str,

    zone_id: str | None,

    device: Device | None = None,

) -> SRMNetworkCapabilityActivateCommand:

    source_filter = request.sourceTrafficFilters

    destination_filter = request.destinationTrafficFilters
@@ -45,7 +62,18 @@ def build_activate_command( 
        application_reference=NetworkCapabilityTargetApplicationReference(

            service_specification_ref=str(request.appId),

            external_app_id=str(request.appId),

            app_instance_id=str(request.appInstanceId) if request.appInstanceId else None,

        ),

        device=NetworkCapabilityTargetDevice(

            phone_number=device.phoneNumber,

            ipv4=device.ipv4Address.publicAddress if device.ipv4Address else None,

            ipv4_port=device.ipv4Address.publicPort if device.ipv4Address else None,

            ipv4_private=device.ipv4Address.privateAddress if device.ipv4Address else None,

            ipv6=device.ipv6Address,

            network_access_id=device.networkAccessIdentifier,

        )

        if device is not None

        else None,

    )



    return SRMNetworkCapabilityActivateCommand(
@@ -161,6 +189,10 @@ def build_traffic_influence_changed_event( 
    data = response.model_dump(mode="json", exclude_none=True)

    if selected_app_instance_ids:

        data["selected_appInstanceId"] = [str(i) for i in selected_app_instance_ids]

    if request_metadata.get("device") is not None:

        device_response = build_device_response(Device.model_validate(request_metadata["device"]))

        if device_response is not None:

            data["deviceResponse"] = device_response.model_dump(mode="json", exclude_none=True)

    return TrafficInfluenceChangedCloudEvent(

        id=uuid4(),

        source=source,

src/open_exposure_gateway/application/services/network_capability_condition.py

0 → 100644
+25 −0
Changes for src/open_exposure_gateway/application/services/network_capability_condition.py: 25 added lines, 0 removed lines.
Original line number Diff line number Diff line 
"""Shared parsing of async network capability condition metadata on status events.



SRM emits ``capability_status`` / ``capability_status_info`` (interface-contract §C.4).

"""



from __future__ import annotations



from typing import Any, Literal, TypeGuard



CapabilityStatus = Literal["AVAILABLE", "UNAVAILABLE"]





def capability_condition(

    metadata: dict[str, Any] | None,

) -> tuple[str | None, str | None]:

    md = metadata or {}

    status = md.get("capability_status")

    info = md.get("capability_status_info")

    status_str = status if isinstance(status, str) else None

    info_str = info if isinstance(info, str) else None

    return status_str, info_str





def is_capability_status(status: str | None) -> TypeGuard[CapabilityStatus]:

    return status in ("AVAILABLE", "UNAVAILABLE")

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+12 −27
Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 12 added lines, 27 removed lines.
Original line number Diff line number Diff line
@@ -22,6 +22,10 @@ from open_exposure_gateway.application.mappers.quality_on_demand_mapper import ( 
    session_target_from_capability,

    session_target_from_row,

)

from open_exposure_gateway.application.services.network_capability_condition import (

    capability_condition,

    is_capability_status,

)

from open_exposure_gateway.core.config import (

    DEFAULT_QOD_SERVICE_SPECIFICATION_ID,

    get_settings,
@@ -423,23 +427,11 @@ class QualityOnDemandService: 
            )

        )



    async def handle_completed(self, event: SRMOperationCompleted) -> None:

    async def handle_completed(self, event: SRMOperationCompleted, operation: Operation) -> 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:

            logger.warning(

                "operation_completed_for_unknown_operation", operation_id=event.operation_id

            )

            return

        if operation.operation_type not in (

            OperationType.QOD_SESSION,

            OperationType.QOD_SESSION_DEACTIVATE,

        ):

            return



        operation_id = operation.operation_id

        status = _OPERATION_COMPLETION_STATUS_MAP[event.status]

        result = None

        if status != OperationStatus.FAILED:
@@ -530,15 +522,11 @@ class QualityOnDemandService: 
        )

        await self._qod_session_repo.save(qod_session.model_copy(update={"state": terminal_state}))



    async def handle_status_changed(self, event: SRMOperationStatus) -> None:

    async def handle_status_changed(self, event: SRMOperationStatus, operation: Operation) -> 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.QOD_SESSION:

            return



        operation_id = operation.operation_id

        qod_session = await self._qod_session_repo.get_by_operation_id(operation_id)

        if (

            qod_session is None
@@ -557,10 +545,10 @@ class QualityOnDemandService: 
            )

            return



        qos_status = (event.metadata or {}).get("qos_status")

        if qos_status not in ("AVAILABLE", "UNAVAILABLE"):

        qos_status, reason = capability_condition(event.metadata)

        if not is_capability_status(qos_status):

            logger.warning(

                "operation_status_unrecognized_qos_status",

                "operation_status_unrecognized_capability_status",

                operation_id=event.operation_id,

                metadata=event.metadata,

            )
@@ -571,10 +559,7 @@ class QualityOnDemandService: 
        )

        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"

            )

            status_info = _QOS_STATUS_INFO_MAP.get(reason or "", "NETWORK_TERMINATED")



        prior_state = qod_session.state

        update: dict[str, object] = {"state": status}