feat: enhance traffic influence handling by excluding deleted states and...

from open_exposure_gateway.adapters.database.mappers import TrafficInfluenceMapper

from open_exposure_gateway.adapters.database.sql import TrafficInfluenceRow

from open_exposure_gateway.domain.models import TrafficInfluence

from open_exposure_gateway.domain.models.traffic_influences.enums import TrafficInfluenceState

from open_exposure_gateway.ports.database.traffic_influences import TrafficInfluenceRepository

    async def list_by_app_registration_id(

        self, app_registration_id: UUID | None

    ) -> list[TrafficInfluence]:

        stmt = select(TrafficInfluenceRow)

        stmt = select(TrafficInfluenceRow).where(

            TrafficInfluenceRow.state != TrafficInfluenceState.DELETED

        )

        if app_registration_id is not None:

            stmt = stmt.where(TrafficInfluenceRow.app_registration_id == app_registration_id)

        rows = await self._session.scalars(stmt)

    tags=["Traffic Influence API read"],

    summary="Retrieves existing TrafficInfluence Resources",

    response_model=list[TrafficInfluence],

    # The CAMARA schema marks no field nullable, so an unset optional must be omitted

    # rather than serialized as null -- same rule the POST response already follows.

    response_model_exclude_none=True,

    responses=_responses(400, 401, 403, 500, 503),

)

async def get_all_traffic_influences(

    summary="Reads a specific TrafficInfluence resource identified by the "

    "trafficInfluenceID value.",

    response_model=TrafficInfluence,

    response_model_exclude_none=True,

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

)

async def get_traffic_influence_by_id(

        if traffic_influence is None:

            raise NotFoundException(message=f"Traffic influence {traffic_influence_id} not found")

        # Best-effort: its read endpoint is still unconfirmed (srm_client.py), so a

        # failure here must never block a GET that OEG's own bookkeeping can already

        # answer -- including the 'deletion in progress'/'deleted' terminal states, where

        # SRM may have already purged the capability_instance entirely. When it does

        # succeed, build_traffic_influence_response translates it back to CAMARA shape

        # the same way build_session_info does for QoD.

        # SRM only holds a capability_instance while the policy is realized, so a 404 is

        # expected unless OEG believes it is ACTIVE: before activation completes there is

        # nothing yet, and once teardown starts SRM may have purged it (TI's CAMARA spec

        # keeps 'deletion in progress'/'deleted' observable, so the read still has to

        # answer). A 404 while ACTIVE means SRM lost a live policy -- that surfaces.

        # DownstreamServiceException always surfaces: answering 200 from stale local

        # bookkeeping turns an SRM outage into a silently thinner response, where a

        # field only SRM knows (edgeCloudZoneId for a region-pinned request) reads as

        # null and the caller cannot tell "not assigned yet" from "could not find out".

        capability = None

        try:

            capability = await self.srm_client.get_traffic_influence(

                traffic_influence_id=traffic_influence_id,

                x_correlator=x_correlator,

            )

        except (NotFoundException, DownstreamServiceException, ValidationError):

            logger.warning(

                "srm_traffic_influence_read_failed", traffic_influence_id=traffic_influence_id

        except NotFoundException:

            if traffic_influence.state == RecordTrafficInfluenceState.ACTIVE:

                raise

            logger.info(

                "srm_traffic_influence_absent",

                traffic_influence_id=traffic_influence_id,

                state=traffic_influence.state.value,

            )

        request_metadata = await self._fetch_request_metadata(traffic_influence.operation_id)

                x_correlator=x_correlator,

            )

            capabilities_by_id = {c.service_instance_id: c for c in capabilities}

        except (NotFoundException, DownstreamServiceException, ValidationError):

            logger.warning("srm_traffic_influence_list_read_failed", app_id=str(app_id))

        except NotFoundException:

            logger.info("srm_traffic_influence_list_absent", app_id=str(app_id))

        responses = []

        for record in records:

    CallbackRegistration,

    Operation,

    QodSession,

    TrafficInfluenceState,

)

from open_exposure_gateway.domain.models import (

    TrafficInfluence as TrafficInfluenceRecord,

        return [

            row.model_copy(deep=True)

            for row in self.rows.values()

            if app_registration_id is None or row.app_registration_id == app_registration_id

            if row.state != TrafficInfluenceState.DELETED

            and (app_registration_id is None or row.app_registration_id == app_registration_id)

        ]

    async def save(self, traffic_influence: TrafficInfluenceRecord) -> TrafficInfluenceRecord:

"""Traffic Influence responses must validate against the vendored CAMARA schema.

The flow tests assert behaviour; this asserts *shape* against traffic_influence.yaml

itself, so a response that drifts from the contract fails here rather than in a

consumer. Notably the spec marks no field `nullable`, so an unset optional must be

omitted, not serialized as null (OpenAPI 3.0.3 semantics).

"""

from pathlib import Path

from typing import Any

from uuid import UUID, uuid4

import pytest

import yaml

from fastapi.testclient import TestClient

from jsonschema import Draft202012Validator

from open_exposure_gateway.api.camara.traffic_influence.vwip.router import BASE_PATH as TI_BASE

from open_exposure_gateway.domain.traffic_influence import Subject

from tests.unit.fakes import (

    FakeDataBus,

    FakeTrafficInfluenceRepository,

    completion_payload,

)

# `registered_app` is imported for its autouse effect in this module: every Traffic

# Influence request targets an already-registered app (ADR-0011).

from tests.unit.test_traffic_influence_flows import TI_BODY, registered_app  # noqa: F401

_SPEC = yaml.safe_load(

    (

        Path(__file__).parents[2]

        / "src/open_exposure_gateway/api/camara/traffic_influence/vwip"

        / "API_definitions/traffic_influence.yaml"

    ).read_text()

)

def _assert_matches(payload: Any, schema_name: str) -> None:

    # Whole document as root so internal $refs (Port, AppId, ...) resolve.

    validator = Draft202012Validator({**_SPEC, "$ref": f"#/components/schemas/{schema_name}"})

    errors = [e.message for e in validator.iter_errors(payload)]

    assert not errors, f"{schema_name} response violates the CAMARA schema: {errors}"

class TestTrafficInfluenceCamaraSchema:

    def test_create_response_matches_schema(self, api_client: TestClient) -> None:

        response = api_client.post(f"{TI_BASE}/traffic-influences", json=TI_BODY)

        assert response.status_code == 201

        _assert_matches(response.json(), "TrafficInfluence")

    def test_get_response_omits_unset_optionals_rather_than_nulling_them(

        self, api_client: TestClient

    ) -> None:

        """The regression this file exists for: `appInstanceId: null` is not valid against

        a schema that never declares a field nullable."""

        body = {k: v for k, v in TI_BODY.items() if k != "edgeCloudZoneId"}

        traffic_influence_id = api_client.post(f"{TI_BASE}/traffic-influences", json=body).json()[

            "trafficInfluenceID"

        ]

        payload = api_client.get(f"{TI_BASE}/traffic-influences/{traffic_influence_id}").json()

        assert "appInstanceId" not in payload

        _assert_matches(payload, "TrafficInfluence")

    def test_collection_response_matches_schema(self, api_client: TestClient) -> None:

        api_client.post(f"{TI_BASE}/traffic-influences", json=TI_BODY)

        for item in api_client.get(f"{TI_BASE}/traffic-influences").json():

            _assert_matches(item, "TrafficInfluence")

    @pytest.mark.usefixtures("live_ti")

    async def test_deleted_response_matches_schema(

        self,

        api_client: TestClient,

        fake_bus: FakeDataBus,

        traffic_influence_repo: FakeTrafficInfluenceRepository,

    ) -> None:

        """'deleted' is a documented value of the CAMARA state enum, so the terminal read

        is a schema-valid 200 -- not a 404."""

        traffic_influence_id = api_client.post(

            f"{TI_BASE}/traffic-influences", json=TI_BODY

        ).json()["trafficInfluenceID"]

        api_client.delete(f"{TI_BASE}/traffic-influences/{traffic_influence_id}")

        deactivate_operation_id = [

            p for s, p in fake_bus.published if s == Subject.TASK_DEACTIVATE

        ][0]["operation_id"]

        await fake_bus.publish(

            str(Subject.OPERATION_COMPLETED),

            completion_payload(

                deactivate_operation_id,

                instances=[

                    {

                        "service_instance_id": str(uuid4()),

                        "zone_id": str(uuid4()),

                        "status": "completed",

                        "external_ref": "traffic-policy-456",

                    }

                ],

            ),

        )

        assert traffic_influence_repo.rows[UUID(traffic_influence_id)].state.value == "deleted"

        response = api_client.get(f"{TI_BASE}/traffic-influences/{traffic_influence_id}")

        assert response.status_code == 200

        assert response.json()["state"] == "deleted"

        _assert_matches(response.json(), "TrafficInfluence")