Merge branch 'fix/federation-schemas-match-gsma' into 'develop'

| **Variable**                | **Description**                                             |

|-----------------------------|-------------------------------------------------------------|

| `SRM_HOST`                  | Base URL of the Service Resource Manager                   |

| `FM_SETTINGS__BASE_URL`     | Base URL of the Federation Manager (default `http://localhost:8082`) |

| `FM_SETTINGS__BASE_URL`     | Federation Manager's internal API, on its internal port (default `http://localhost:8083`) |

| `FM_SETTINGS__TIMEOUT`      | Seconds to wait for Federation Manager (default `30`) |

    FAULT_MGMT = "faultMgmt"

    EVENT_MGMT = "eventMgmt"

    RESOURCE_MONITOR = "resourceMonitor"

    NETWORK_EVENT_MGMT = "networkEventMgmt"

    APP_NOTIFICATION_MGMT = "appNotificationMgmt"

    APP_LEVEL_POLICY_MGMT = "appLevelPolicyMgmt"

    OPS_LEVEL_POLICY_MGMT = "opsLevelPolicyMgmt"

# --- Object schemas --------------------------------------------------------------

    """An availability zone an OP offers to the developers of a federation partner."""

    zoneId: ZoneIdentifier

    geolocation: GeoLocation

    geolocation: Optional[GeoLocation] = None

    geographyDetails: str

class FederationDetails(BaseModel):

    """``200`` body of ``GetFederationDetails`` / ``UpdateFederation``."""

    """``200`` body of ``GetFederationDetails`` / ``UpdateFederation``. OPG.04 makes every

    field optional."""

    edgeDiscoveryServiceEndPoint: ServiceEndpoint

    lcmServiceEndPoint: ServiceEndpoint

    edgeDiscoveryServiceEndPoint: Optional[ServiceEndpoint] = None

    lcmServiceEndPoint: Optional[ServiceEndpoint] = None

    allowedMobileNetworkIds: Optional[MobileNetworkIds] = None

    allowedFixedNetworkIds: Optional[FixedNetworkIds] = None

    offeredAvailabilityZones: Optional[Annotated[list[ZoneDetails], Field(min_length=1)]] = None

    platformCaps: Optional[list[PlatformCapability]] = None

class FederationContextIdResponse(BaseModel):

class FmSettings(BaseModel):

    base_url: HttpUrl = HttpUrl("http://localhost:8082")

    timeout: float = 10.0

    # FM's internal listener; it doesn't serve /internal/* on its EWBI port

    base_url: HttpUrl = HttpUrl("http://localhost:8083")

    # FM may fetch a token from the partner's Keycloak and then call the partner, with up

    # to 10s for each. A shorter wait gives up while FM still finishes the operation, so

    # the operator sees a 503 for a federation that was in fact created.

    timeout: float = 30.0

class ReconciliationSettings(BaseModel):

"""OEG's federation models must accept everything the bundled OPG.04 profile allows.

OEG checks FM's answers against these models and turns a mismatch into a 502, so a model

stricter than the spec rejects partners that follow it. This walks each relayed

operation's request and response schema next to the model's JSON schema, field by field

and through nested objects and arrays, and reports where the model asks for more:

a field it requires that the spec leaves optional, a spec field it lacks, or an enum

value it doesn't know.

"""

from pathlib import Path

from typing import Any

import pytest

import yaml

from pydantic import TypeAdapter

from open_exposure_gateway.api.gsma.federation_manager.v1_4_0 import schemas

_SPEC: dict[str, Any] = yaml.safe_load(

    (

        Path(__file__).parents[2]

        / "src/open_exposure_gateway/api/gsma/federation_manager/API_definitions"

        / "OPG.04-v6.0-EWBI-Federation-API-v1.4.0-oop-profile.yaml"

    ).read_text()

)

# GSMA operationId -> (request model, 200 response model) of each operation OEG relays.

_OPERATIONS: dict[str, tuple[Any, Any]] = {

    "CreateFederation": (schemas.FederationRequestData, schemas.FederationResponseData),

    "GetFederationContextId": (None, schemas.FederationContextIdResponse),

    "GetFederationDetails": (None, schemas.FederationDetails),

    "GetZoneData": (None, list[schemas.ZoneRegisteredData]),

    "ZoneSubscribe": (schemas.ZoneRegistrationRequestData, schemas.ZoneRegistrationResponseData),

    "GetZoneDetails": (None, schemas.ZoneRegisteredData),

}

def _deref(schema: dict[str, Any], root: dict[str, Any]) -> dict[str, Any]:

    while "$ref" in schema:

        node: Any = root

        for part in schema["$ref"].removeprefix("#/").split("/"):

            node = node[part]

        schema = node

    return schema

def _unwrap(schema: dict[str, Any], root: dict[str, Any]) -> dict[str, Any]:

    """The schema itself, past $refs and the anyOf-with-null pydantic writes for Optional."""

    schema = _deref(schema, root)

    for key in ("anyOf", "oneOf"):

        if key in schema:

            options = [_deref(o, root) for o in schema[key]]

            options = [o for o in options if o.get("type") != "null"]

            if len(options) == 1:

                return _unwrap(options[0], root)

    if len(schema.get("allOf", [])) == 1:

        return _unwrap(schema["allOf"][0], root)

    return schema

def _stricter(

    spec: dict[str, Any], model: dict[str, Any], model_root: dict[str, Any], where: str

) -> list[str]:

    spec, model = _unwrap(spec, _SPEC), _unwrap(model, model_root)

    found = []

    if "enum" in spec and "enum" in model:

        unknown = set(spec["enum"]) - set(model["enum"])

        if unknown:

            found.append(f"{where}: rejects spec values {sorted(unknown)}")

    if "items" in spec and "items" in model:

        found += _stricter(spec["items"], model["items"], model_root, f"{where}[]")

    if "properties" in spec and "properties" in model:

        required = set(model.get("required", [])) - set(spec.get("required", []))

        if required:

            found.append(f"{where}: requires {sorted(required)}, which the spec leaves optional")

        missing = set(spec["properties"]) - set(model["properties"])

        if missing:

            found.append(f"{where}: lacks spec fields {sorted(missing)}")

        for name in sorted(set(spec["properties"]) & set(model["properties"])):

            found += _stricter(

                spec["properties"][name], model["properties"][name], model_root, f"{where}.{name}"

            )

    return found

def _operation(operation_id: str) -> dict[str, Any]:

    for item in _SPEC["paths"].values():

        for operation in item.values():

            if isinstance(operation, dict) and operation.get("operationId") == operation_id:

                return operation

    raise AssertionError(f"{operation_id} is not in the bundled spec")

def _body(operation_id: str, direction: str) -> dict[str, Any]:

    operation = _operation(operation_id)

    if direction == "request":

        content = _deref(operation["requestBody"], _SPEC)["content"]

    else:

        content = _deref(operation["responses"]["200"], _SPEC)["content"]

    schema: dict[str, Any] = content["application/json"]["schema"]

    return schema

_CASES = [

    pytest.param(operation_id, direction, model, id=f"{operation_id}-{direction}")

    for operation_id, (request, response) in _OPERATIONS.items()

    for direction, model in (("request", request), ("response", response))

    if model is not None

]

@pytest.mark.parametrize(("operation_id", "direction", "model"), _CASES)

def test_model_accepts_everything_the_spec_allows(

    operation_id: str, direction: str, model: Any

) -> None:

    model_schema = TypeAdapter(model).json_schema(ref_template="#/$defs/{model}")

    found = _stricter(_body(operation_id, direction), model_schema, model_schema, operation_id)

    assert not found, "\n".join(found)

def test_the_walk_catches_a_model_stricter_than_the_spec() -> None:

    """Guards the walk itself: it must flag a field the spec leaves optional."""

    spec = {"type": "object", "properties": {"a": {"type": "string"}}}

    model = {"type": "object", "properties": {"a": {"type": "string"}}, "required": ["a"]}

    assert _stricter(spec, model, model, "X") == [

        "X: requires ['a'], which the spec leaves optional"

    ]

    assert response.content == content

# OPG.04 makes every field of GetFederationDetails' answer optional, so an empty or

# unrelated object is a valid one there; it still breaks every other operation's schema.

_OBJECT_BAD_BODIES = (b"{}", b'{"unrelated": true}')

_DETAILS_URL = f"{BASE_PATH}{_FEDERATIONS}/{_CTX}"

@pytest.mark.parametrize(

    ("http_method", "url", "fm_method", "fm_path", "body", "fm_body"),

    [e for e in _ENDPOINTS if e[5] is not None],

)

@pytest.mark.parametrize(

    "bad_body",

    [b"", b"{}", b"[]", b'{"unrelated": true}', b"{not valid json"],

    ("http_method", "url", "fm_method", "fm_path", "body", "fm_body", "bad_body"),

    [

        (*e, bad)

        for e in _ENDPOINTS

        if e[5] is not None

        for bad in (b"", b"[]", b"{not valid json", *_OBJECT_BAD_BODIES)

        if not (e[1] == _DETAILS_URL and e[0] == "get" and bad in _OBJECT_BAD_BODIES)

    ],

)

def test_fm_success_body_not_matching_the_response_schema_returns_502(

    client: TestClient,

    _assert_bad_response_502(client.request(http_method, url, content=body, headers=headers))

def test_fm_success_body_for_another_operation_returns_502(

def test_fm_details_with_a_field_of_the_wrong_type_returns_502(

    client: TestClient, fake_fm_client: _FakeFmClient

) -> None:

    """A valid GSMA body, just not this operation's: GetFederationDetails answered with

    CreateFederation's response is still outside the contract."""

    fake_fm_client.response = _fm_ok(_FEDERATION_RESPONSE)

    """Every field is optional, but one that's there must still have the spec's type."""

    fake_fm_client.response = _fm_ok(b'{"platformCaps": "homeRouting"}')

    _assert_bad_response_502(client.get(f"{BASE_PATH}{_FEDERATIONS}/{_CTX}"))

    _assert_bad_response_502(client.get(_DETAILS_URL))

@pytest.mark.parametrize(

    "content",

    [

        pytest.param(b"{}", id="no-fields"),

        pytest.param(

            b'{"allowedMobileNetworkIds": {"mcc": "208", "mncs": ["01"]}, '

            b'"platformCaps": ["serviceAPIs", "networkEventMgmt"], '

            b'"offeredAvailabilityZones": [{"zoneId": "zone-a", "geographyDetails": "Paris"}]}',

            id="no-endpoints-and-a-zone-without-geolocation",

        ),

    ],

)

def test_fm_details_the_spec_allows_are_relayed_byte_for_byte(

    client: TestClient, fake_fm_client: _FakeFmClient, content: bytes

) -> None:

    """What FM sends for a partner that publishes no service endpoints: OPG.04 makes them,

    and a zone's geolocation, optional."""

    fake_fm_client.response = _fm_ok(content)

    response = client.get(_DETAILS_URL)

    assert response.status_code == 200

    assert response.content == content

def test_fm_create_answer_with_any_spec_platform_capability_is_relayed(

    client: TestClient, fake_fm_client: _FakeFmClient

) -> None:

    content = (

        b'{"federationContextId": "fed-ctx-1", "platformCaps": ["homeRouting", "faultMgmt", '

        b'"networkEventMgmt", "appNotificationMgmt", "appLevelPolicyMgmt", "opsLevelPolicyMgmt"]}'

    )

    fake_fm_client.response = _fm_ok(content)

    response = client.post(

        f"{BASE_PATH}{_FEDERATIONS}",

        content=_VALID_CREATE_BODY,

        headers={"content-type": "application/json"},

    )

    assert response.status_code == 200

    assert response.content == content

def test_fm_success_body_with_unknown_fields_is_relayed_byte_for_byte(