Loading src/open_exposure_gateway/api/gsma/federation_manager/v1_2_0/router.py +99 −0 Original line number Diff line number Diff line """FastAPI routes for the GSMA Federation Manager API (EWBI OPG v1.2.0). Endpoints defined by ``API_definitions/federation-manager.yaml``. This module currently exposes the ``FederationManagement`` tag only, and every handler is a surface stub that raises :class:`NotImplementedException` (HTTP 501). The request/response contract (path, schema, error codes) is live and visible in the OpenAPI document; the fulfilment logic is added in a later step. """ from typing import Any from fastapi import APIRouter from open_exposure_gateway.api.gsma.federation_manager.v1_2_0.schemas import ( FederationContextId, FederationContextIdResponse, FederationDetails, FederationPatchRequest, FederationRequestData, FederationResponseData, ProblemDetails, ) from open_exposure_gateway.core.exceptions import NotImplementedException # GSMA OPG serves the Federation Management API at {apiRoot}/operatorplatform/federation/v1. BASE_PATH = "/operatorplatform/federation/v1" router = APIRouter(prefix=BASE_PATH, tags=["Federation Manager"]) # Every Federation Manager error response is an RFC 7807 ProblemDetails (federation-manager.yaml). _ERROR_RESPONSES: dict[int | str, dict[str, Any]] = { 400: {"model": ProblemDetails, "description": "Bad request"}, 401: {"model": ProblemDetails, "description": "Unauthorized"}, 404: {"model": ProblemDetails, "description": "Not found"}, 409: {"model": ProblemDetails, "description": "Conflict"}, 422: {"model": ProblemDetails, "description": "Unprocessable entity"}, 500: {"model": ProblemDetails, "description": "Internal server error"}, 501: {"model": ProblemDetails, "description": "Not implemented"}, 503: {"model": ProblemDetails, "description": "Service unavailable"}, 520: {"model": ProblemDetails, "description": "Unknown error"}, } def _responses(*codes: int) -> dict[int | str, dict[str, Any]]: return {code: _ERROR_RESPONSES[code] for code in codes} _NOT_IMPLEMENTED = "Federation Manager is not implemented in this release" @router.post( "/partner", summary="Create a one-direction federation with a partner operator platform", operation_id="create_federation", response_model=FederationResponseData, response_model_exclude_none=True, responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520), ) async def create_federation(request: FederationRequestData) -> Any: raise NotImplementedException(message=_NOT_IMPLEMENTED) @router.get( "/{federationContextId}/partner", summary="Retrieve details about the federation context with the partner OP", operation_id="get_federation_details", response_model=FederationDetails, response_model_exclude_none=True, responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520), ) async def get_federation_details(federationContextId: FederationContextId) -> Any: raise NotImplementedException(message=_NOT_IMPLEMENTED) @router.patch( "/{federationContextId}/partner", summary="Update the parameters associated with an existing federation", operation_id="update_federation", response_model=FederationDetails, response_model_exclude_none=True, responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520), ) async def update_federation( federationContextId: FederationContextId, request: FederationPatchRequest ) -> Any: raise NotImplementedException(message=_NOT_IMPLEMENTED) @router.delete( "/{federationContextId}/partner", summary="Remove an existing federation with the partner OP", operation_id="delete_federation_details", status_code=200, responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520), ) async def delete_federation_details(federationContextId: FederationContextId) -> Any: raise NotImplementedException(message=_NOT_IMPLEMENTED) @router.get( "/fed-context-id", summary="Retrieve the existing federationContextId with the partner operator platform", operation_id="get_federation_context_id", response_model=FederationContextIdResponse, response_model_exclude_none=True, responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520), ) async def get_federation_context_id() -> Any: raise NotImplementedException(message=_NOT_IMPLEMENTED) src/open_exposure_gateway/api/gsma/federation_manager/v1_2_0/schemas.py +219 −1 Original line number Diff line number Diff line """Pydantic models for the GSMA Federation Manager API (EWBI OPG v1.2.0). Request/response schemas derived from ``API_definitions/federation-manager.yaml``. ``API_definitions/federation-manager.yaml``. Only the ``FederationManagement`` tag is modelled here; the other tags (zones, artefacts, application onboarding and LCM) are added as they are implemented. Schema names follow the OpenAPI ``operationId`` intent rather than the generated ``inline_response_*`` / ``*_body`` names: ``FederationDetails`` is the spec's ``inline_response_200_1``, ``FederationContextIdResponse`` is ``inline_response_200_2`` and ``FederationPatchRequest`` is ``federationContextId_partner_body``. """ from __future__ import annotations from datetime import datetime from enum import StrEnum from typing import Annotated, Optional from pydantic import BaseModel, ConfigDict, Field, StringConstraints, model_validator # --- Scalar types (OpenAPI: string/integer with pattern or format) ------------- # "^[A-Za-z0-9][A-Za-z0-9-]*$" is shared by FederationContextId, FederationIdentifier # and ZoneIdentifier in the spec. _ID_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9-]*$" FederationContextId = Annotated[str, StringConstraints(pattern=_ID_PATTERN)] """Allocated by the partner OP when a federation context is created; echoed by the Originating OP on every subsequent request.""" FederationIdentifier = Annotated[str, StringConstraints(pattern=_ID_PATTERN)] """Globally unique identifier of an operator platform, in the MEC federation context.""" ZoneIdentifier = Annotated[str, StringConstraints(pattern=_ID_PATTERN)] CountryCode = Annotated[str, StringConstraints(pattern=r"^[A-Z]{2}$")] """ISO 3166-1 Alpha-2 country code of the partner operator.""" Mcc = Annotated[str, StringConstraints(pattern=r"^\d{3}$")] Mnc = Annotated[str, StringConstraints(pattern=r"^\d{2,3}$")] GeoLocation = Annotated[ str, StringConstraints( pattern=r"^([-+]?)([\d]{1,2})((((\.)([\d]{1,4}))?(,)))(([-+]?)([\d]{1,3})((\.)([\d]{1,4}))?)$" ), ] """``Latitude,Longitude`` as a decimal fraction, up to 4-digit precision.""" Ipv4Addr = Annotated[ str, StringConstraints( pattern=r"^(([0-9]|[1-9][0-9]|1[0-9][0-9]|2[0-4][0-9]|25[0-5])\.){3}" r"([0-9]|[1-9][0-9]|1[0-9][0-9]|2[0-4][0-9]|25[0-5])$" ), ] Ipv6Addr = Annotated[ str, StringConstraints( pattern=r"^((:|(0?|([1-9a-f][0-9a-f]{0,3}))):)" r"((0?|([1-9a-f][0-9a-f]{0,3})):){0,6}(:|(0?|([1-9a-f][0-9a-f]{0,3})))$" ), ] Fqdn = str Uri = str Port = Annotated[int, Field(ge=0)] # A non-empty list of fixed-line network identifiers. FixedNetworkIds = Annotated[list[str], Field(min_length=1)] # --- Enumerations ------------------------------------------------------------- class PlatformCapability(StrEnum): """Capabilities an operator platform advertises to a federation partner.""" HOME_ROUTING = "homeRouting" ANCHORING = "Anchoring" SERVICE_APIS = "serviceAPIs" FAULT_MGMT = "faultMgmt" EVENT_MGMT = "eventMgmt" RESOURCE_MONITOR = "resourceMonitor" class NetworkCodeObjectType(StrEnum): MOBILE_NETWORK_CODES = "MOBILE_NETWORK_CODES" FIXED_NETWORK_CODES = "FIXED_NETWORK_CODES" class NetworkCodeOperationType(StrEnum): ADD_CODES = "ADD_CODES" REMOVE_CODES = "REMOVE_CODES" UPDATE_CODES = "UPDATE_CODES" # --- Object schemas -------------------------------------------------------------- class MobileNetworkIds(BaseModel): """MCC plus the set of MNCs associated with an operator platform's mobile network.""" mcc: Optional[Mcc] = None mncs: Optional[Annotated[list[Mnc], Field(min_length=1)]] = None class CallbackCredentials(BaseModel): """OAuth2 client-credentials used by the partner OP to authenticate callbacks.""" model_config = ConfigDict(extra="forbid") tokenUrl: Uri clientId: str clientSecret: str class ServiceEndpoint(BaseModel): """Reachability information for the edge-discovery or LCM service of an OP. The spec requires ``port`` plus at least one of ``fqdn`` / ``ipv4Addresses`` / ``ipv6Addresses``. """ port: Port fqdn: Optional[Fqdn] = None ipv4Addresses: Optional[Annotated[list[Ipv4Addr], Field(min_length=1)]] = None ipv6Addresses: Optional[Annotated[list[Ipv6Addr], Field(min_length=1)]] = None @model_validator(mode="after") def _at_least_one_address(self) -> ServiceEndpoint: if not (self.fqdn or self.ipv4Addresses or self.ipv6Addresses): raise ValueError("one of fqdn, ipv4Addresses or ipv6Addresses is required") return self class ZoneDetails(BaseModel): """An availability zone an OP offers to the developers of a federation partner.""" zoneId: ZoneIdentifier geolocation: GeoLocation geographyDetails: str class InvalidParam(BaseModel): param: str reason: Optional[str] = None class ProblemDetails(BaseModel): """RFC 7807 error body used by every Federation Manager error response.""" title: Optional[str] = None detail: Optional[str] = None cause: Optional[str] = None invalidParams: Optional[Annotated[list[InvalidParam], Field(min_length=1)]] = None class FederationRequestData(BaseModel): """Body of ``POST /partner`` -- the Originating OP's federation create request.""" model_config = ConfigDict(extra="forbid") origOPFederationId: FederationIdentifier initialDate: datetime partnerStatusLink: Uri origOPCountryCode: Optional[CountryCode] = None origOPMobileNetworkCodes: Optional[MobileNetworkIds] = None origOPFixedNetworkCodes: Optional[FixedNetworkIds] = None partnerCallbackCredentials: Optional[CallbackCredentials] = None class FederationResponseData(BaseModel): """``200`` body of ``POST /partner`` -- the partner OP's federation context.""" partnerOPFederationId: FederationIdentifier federationContextId: FederationContextId platformCaps: list[PlatformCapability] partnerOPCountryCode: Optional[CountryCode] = None edgeDiscoveryServiceEndPoint: Optional[ServiceEndpoint] = None lcmServiceEndPoint: Optional[ServiceEndpoint] = None partnerOPMobileNetworkCodes: Optional[MobileNetworkIds] = None partnerOPFixedNetworkCodes: Optional[FixedNetworkIds] = None offeredAvailabilityZones: Optional[Annotated[list[ZoneDetails], Field(min_length=1)]] = None class FederationDetails(BaseModel): """``200`` body of ``GET`` / ``PATCH`` ``/{federationContextId}/partner``. The spec's ``inline_response_200_1``. """ edgeDiscoveryServiceEndPoint: ServiceEndpoint lcmServiceEndPoint: ServiceEndpoint allowedMobileNetworkIds: Optional[MobileNetworkIds] = None allowedFixedNetworkIds: Optional[FixedNetworkIds] = None offeredAvailabilityZones: Optional[Annotated[list[ZoneDetails], Field(min_length=1)]] = None class FederationContextIdResponse(BaseModel): """``200`` body of ``GET /fed-context-id``. The spec's ``inline_response_200_2``. The wire field name is ``FederationContextId`` (capitalised) per the spec. """ FederationContextId: FederationContextId class FederationPatchRequest(BaseModel): """Body of ``PATCH /{federationContextId}/partner``. The spec's ``federationContextId_partner_body``. """ model_config = ConfigDict(extra="forbid") objectType: NetworkCodeObjectType operationType: NetworkCodeOperationType modificationDate: datetime addMobileNetworkIds: Optional[MobileNetworkIds] = None removeMobileNetworkIds: Optional[MobileNetworkIds] = None addFixedNetworkIds: Optional[FixedNetworkIds] = None removeFixedNetworkIds: Optional[FixedNetworkIds] = None src/open_exposure_gateway/main.py +8 −0 Original line number Diff line number Diff line Loading @@ -44,6 +44,9 @@ from open_exposure_gateway.api.error_handlers import ( register_exception_handlers, x_correlator_header, ) from open_exposure_gateway.api.gsma.federation_manager.v1_2_0.router import ( router as federation_manager_router, ) from open_exposure_gateway.api.platform.health import router as health_router from open_exposure_gateway.application.services.edge_application_management_service import ( EdgeApplicationManagementService, Loading Loading @@ -159,6 +162,10 @@ openapi_tags = [ "name": "Location retrieval", "description": "Retrieve the location of a device", }, { "name": "Federation Manager", "description": "GSMA OPG Federation Management (EWBI) -- partner federation lifecycle", }, { "name": "Platform", "description": "Platform-specific endpoints (health, readiness probes)", Loading Loading @@ -309,6 +316,7 @@ def create_app(lifespan: Optional[Lifespan[FastAPI]] = None) -> FastAPI: app.include_router(edge_application_management_router) app.include_router(quality_on_demand_router) app.include_router(location_retrieval_router) app.include_router(federation_manager_router) return app Loading tests/unit/test_federation_manager_endpoints.py 0 → 100644 +77 −0 Original line number Diff line number Diff line """GSMA Federation Manager -- surface stub. The FederationManagement routes are wired into the app and visible in the OpenAPI document, but every handler returns HTTP 501 until fulfilment logic is added. """ from datetime import datetime, timezone from typing import Any import pytest from fastapi.testclient import TestClient from pydantic import ValidationError from open_exposure_gateway.api.gsma.federation_manager.v1_2_0.router import BASE_PATH from open_exposure_gateway.api.gsma.federation_manager.v1_2_0.schemas import ( FederationRequestData, ) from open_exposure_gateway.main import app client = TestClient(app, raise_server_exceptions=False) _CTX = "fed-ctx-1" _VALID_CREATE = { "origOPFederationId": "orig-op-1", "initialDate": "2026-09-04T12:00:00Z", "partnerStatusLink": "https://orig-op.example.com/status", } _VALID_PATCH = { "objectType": "MOBILE_NETWORK_CODES", "operationType": "ADD_CODES", "modificationDate": "2026-09-04T12:00:00Z", } _ENDPOINTS = [ ("post", f"{BASE_PATH}/partner", _VALID_CREATE), ("get", f"{BASE_PATH}/{_CTX}/partner", None), ("patch", f"{BASE_PATH}/{_CTX}/partner", _VALID_PATCH), ("delete", f"{BASE_PATH}/{_CTX}/partner", None), ("get", f"{BASE_PATH}/fed-context-id", None), ] _OPERATION_IDS = { "create_federation", "get_federation_details", "update_federation", "delete_federation_details", "get_federation_context_id", } @pytest.mark.parametrize(("method", "path", "body"), _ENDPOINTS) def test_endpoint_returns_501(method: str, path: str, body: dict[str, Any] | None) -> None: response = client.request(method, path, json=body) assert response.status_code == 501 def test_openapi_exposes_federation_management_operations() -> None: paths = app.openapi()["paths"] seen = { op["operationId"] for methods in paths.values() for op in methods.values() if isinstance(op, dict) and "operationId" in op } assert _OPERATION_IDS <= seen def test_federation_request_data_round_trips() -> None: model = FederationRequestData.model_validate(_VALID_CREATE) assert model.origOPFederationId == "orig-op-1" assert model.initialDate == datetime(2026, 9, 4, 12, 0, tzinfo=timezone.utc) def test_federation_request_data_requires_partner_status_link() -> None: payload = {k: v for k, v in _VALID_CREATE.items() if k != "partnerStatusLink"} with pytest.raises(ValidationError): FederationRequestData.model_validate(payload) Loading
src/open_exposure_gateway/api/gsma/federation_manager/v1_2_0/router.py +99 −0 Original line number Diff line number Diff line """FastAPI routes for the GSMA Federation Manager API (EWBI OPG v1.2.0). Endpoints defined by ``API_definitions/federation-manager.yaml``. This module currently exposes the ``FederationManagement`` tag only, and every handler is a surface stub that raises :class:`NotImplementedException` (HTTP 501). The request/response contract (path, schema, error codes) is live and visible in the OpenAPI document; the fulfilment logic is added in a later step. """ from typing import Any from fastapi import APIRouter from open_exposure_gateway.api.gsma.federation_manager.v1_2_0.schemas import ( FederationContextId, FederationContextIdResponse, FederationDetails, FederationPatchRequest, FederationRequestData, FederationResponseData, ProblemDetails, ) from open_exposure_gateway.core.exceptions import NotImplementedException # GSMA OPG serves the Federation Management API at {apiRoot}/operatorplatform/federation/v1. BASE_PATH = "/operatorplatform/federation/v1" router = APIRouter(prefix=BASE_PATH, tags=["Federation Manager"]) # Every Federation Manager error response is an RFC 7807 ProblemDetails (federation-manager.yaml). _ERROR_RESPONSES: dict[int | str, dict[str, Any]] = { 400: {"model": ProblemDetails, "description": "Bad request"}, 401: {"model": ProblemDetails, "description": "Unauthorized"}, 404: {"model": ProblemDetails, "description": "Not found"}, 409: {"model": ProblemDetails, "description": "Conflict"}, 422: {"model": ProblemDetails, "description": "Unprocessable entity"}, 500: {"model": ProblemDetails, "description": "Internal server error"}, 501: {"model": ProblemDetails, "description": "Not implemented"}, 503: {"model": ProblemDetails, "description": "Service unavailable"}, 520: {"model": ProblemDetails, "description": "Unknown error"}, } def _responses(*codes: int) -> dict[int | str, dict[str, Any]]: return {code: _ERROR_RESPONSES[code] for code in codes} _NOT_IMPLEMENTED = "Federation Manager is not implemented in this release" @router.post( "/partner", summary="Create a one-direction federation with a partner operator platform", operation_id="create_federation", response_model=FederationResponseData, response_model_exclude_none=True, responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520), ) async def create_federation(request: FederationRequestData) -> Any: raise NotImplementedException(message=_NOT_IMPLEMENTED) @router.get( "/{federationContextId}/partner", summary="Retrieve details about the federation context with the partner OP", operation_id="get_federation_details", response_model=FederationDetails, response_model_exclude_none=True, responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520), ) async def get_federation_details(federationContextId: FederationContextId) -> Any: raise NotImplementedException(message=_NOT_IMPLEMENTED) @router.patch( "/{federationContextId}/partner", summary="Update the parameters associated with an existing federation", operation_id="update_federation", response_model=FederationDetails, response_model_exclude_none=True, responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520), ) async def update_federation( federationContextId: FederationContextId, request: FederationPatchRequest ) -> Any: raise NotImplementedException(message=_NOT_IMPLEMENTED) @router.delete( "/{federationContextId}/partner", summary="Remove an existing federation with the partner OP", operation_id="delete_federation_details", status_code=200, responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520), ) async def delete_federation_details(federationContextId: FederationContextId) -> Any: raise NotImplementedException(message=_NOT_IMPLEMENTED) @router.get( "/fed-context-id", summary="Retrieve the existing federationContextId with the partner operator platform", operation_id="get_federation_context_id", response_model=FederationContextIdResponse, response_model_exclude_none=True, responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520), ) async def get_federation_context_id() -> Any: raise NotImplementedException(message=_NOT_IMPLEMENTED)
src/open_exposure_gateway/api/gsma/federation_manager/v1_2_0/schemas.py +219 −1 Original line number Diff line number Diff line """Pydantic models for the GSMA Federation Manager API (EWBI OPG v1.2.0). Request/response schemas derived from ``API_definitions/federation-manager.yaml``. ``API_definitions/federation-manager.yaml``. Only the ``FederationManagement`` tag is modelled here; the other tags (zones, artefacts, application onboarding and LCM) are added as they are implemented. Schema names follow the OpenAPI ``operationId`` intent rather than the generated ``inline_response_*`` / ``*_body`` names: ``FederationDetails`` is the spec's ``inline_response_200_1``, ``FederationContextIdResponse`` is ``inline_response_200_2`` and ``FederationPatchRequest`` is ``federationContextId_partner_body``. """ from __future__ import annotations from datetime import datetime from enum import StrEnum from typing import Annotated, Optional from pydantic import BaseModel, ConfigDict, Field, StringConstraints, model_validator # --- Scalar types (OpenAPI: string/integer with pattern or format) ------------- # "^[A-Za-z0-9][A-Za-z0-9-]*$" is shared by FederationContextId, FederationIdentifier # and ZoneIdentifier in the spec. _ID_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9-]*$" FederationContextId = Annotated[str, StringConstraints(pattern=_ID_PATTERN)] """Allocated by the partner OP when a federation context is created; echoed by the Originating OP on every subsequent request.""" FederationIdentifier = Annotated[str, StringConstraints(pattern=_ID_PATTERN)] """Globally unique identifier of an operator platform, in the MEC federation context.""" ZoneIdentifier = Annotated[str, StringConstraints(pattern=_ID_PATTERN)] CountryCode = Annotated[str, StringConstraints(pattern=r"^[A-Z]{2}$")] """ISO 3166-1 Alpha-2 country code of the partner operator.""" Mcc = Annotated[str, StringConstraints(pattern=r"^\d{3}$")] Mnc = Annotated[str, StringConstraints(pattern=r"^\d{2,3}$")] GeoLocation = Annotated[ str, StringConstraints( pattern=r"^([-+]?)([\d]{1,2})((((\.)([\d]{1,4}))?(,)))(([-+]?)([\d]{1,3})((\.)([\d]{1,4}))?)$" ), ] """``Latitude,Longitude`` as a decimal fraction, up to 4-digit precision.""" Ipv4Addr = Annotated[ str, StringConstraints( pattern=r"^(([0-9]|[1-9][0-9]|1[0-9][0-9]|2[0-4][0-9]|25[0-5])\.){3}" r"([0-9]|[1-9][0-9]|1[0-9][0-9]|2[0-4][0-9]|25[0-5])$" ), ] Ipv6Addr = Annotated[ str, StringConstraints( pattern=r"^((:|(0?|([1-9a-f][0-9a-f]{0,3}))):)" r"((0?|([1-9a-f][0-9a-f]{0,3})):){0,6}(:|(0?|([1-9a-f][0-9a-f]{0,3})))$" ), ] Fqdn = str Uri = str Port = Annotated[int, Field(ge=0)] # A non-empty list of fixed-line network identifiers. FixedNetworkIds = Annotated[list[str], Field(min_length=1)] # --- Enumerations ------------------------------------------------------------- class PlatformCapability(StrEnum): """Capabilities an operator platform advertises to a federation partner.""" HOME_ROUTING = "homeRouting" ANCHORING = "Anchoring" SERVICE_APIS = "serviceAPIs" FAULT_MGMT = "faultMgmt" EVENT_MGMT = "eventMgmt" RESOURCE_MONITOR = "resourceMonitor" class NetworkCodeObjectType(StrEnum): MOBILE_NETWORK_CODES = "MOBILE_NETWORK_CODES" FIXED_NETWORK_CODES = "FIXED_NETWORK_CODES" class NetworkCodeOperationType(StrEnum): ADD_CODES = "ADD_CODES" REMOVE_CODES = "REMOVE_CODES" UPDATE_CODES = "UPDATE_CODES" # --- Object schemas -------------------------------------------------------------- class MobileNetworkIds(BaseModel): """MCC plus the set of MNCs associated with an operator platform's mobile network.""" mcc: Optional[Mcc] = None mncs: Optional[Annotated[list[Mnc], Field(min_length=1)]] = None class CallbackCredentials(BaseModel): """OAuth2 client-credentials used by the partner OP to authenticate callbacks.""" model_config = ConfigDict(extra="forbid") tokenUrl: Uri clientId: str clientSecret: str class ServiceEndpoint(BaseModel): """Reachability information for the edge-discovery or LCM service of an OP. The spec requires ``port`` plus at least one of ``fqdn`` / ``ipv4Addresses`` / ``ipv6Addresses``. """ port: Port fqdn: Optional[Fqdn] = None ipv4Addresses: Optional[Annotated[list[Ipv4Addr], Field(min_length=1)]] = None ipv6Addresses: Optional[Annotated[list[Ipv6Addr], Field(min_length=1)]] = None @model_validator(mode="after") def _at_least_one_address(self) -> ServiceEndpoint: if not (self.fqdn or self.ipv4Addresses or self.ipv6Addresses): raise ValueError("one of fqdn, ipv4Addresses or ipv6Addresses is required") return self class ZoneDetails(BaseModel): """An availability zone an OP offers to the developers of a federation partner.""" zoneId: ZoneIdentifier geolocation: GeoLocation geographyDetails: str class InvalidParam(BaseModel): param: str reason: Optional[str] = None class ProblemDetails(BaseModel): """RFC 7807 error body used by every Federation Manager error response.""" title: Optional[str] = None detail: Optional[str] = None cause: Optional[str] = None invalidParams: Optional[Annotated[list[InvalidParam], Field(min_length=1)]] = None class FederationRequestData(BaseModel): """Body of ``POST /partner`` -- the Originating OP's federation create request.""" model_config = ConfigDict(extra="forbid") origOPFederationId: FederationIdentifier initialDate: datetime partnerStatusLink: Uri origOPCountryCode: Optional[CountryCode] = None origOPMobileNetworkCodes: Optional[MobileNetworkIds] = None origOPFixedNetworkCodes: Optional[FixedNetworkIds] = None partnerCallbackCredentials: Optional[CallbackCredentials] = None class FederationResponseData(BaseModel): """``200`` body of ``POST /partner`` -- the partner OP's federation context.""" partnerOPFederationId: FederationIdentifier federationContextId: FederationContextId platformCaps: list[PlatformCapability] partnerOPCountryCode: Optional[CountryCode] = None edgeDiscoveryServiceEndPoint: Optional[ServiceEndpoint] = None lcmServiceEndPoint: Optional[ServiceEndpoint] = None partnerOPMobileNetworkCodes: Optional[MobileNetworkIds] = None partnerOPFixedNetworkCodes: Optional[FixedNetworkIds] = None offeredAvailabilityZones: Optional[Annotated[list[ZoneDetails], Field(min_length=1)]] = None class FederationDetails(BaseModel): """``200`` body of ``GET`` / ``PATCH`` ``/{federationContextId}/partner``. The spec's ``inline_response_200_1``. """ edgeDiscoveryServiceEndPoint: ServiceEndpoint lcmServiceEndPoint: ServiceEndpoint allowedMobileNetworkIds: Optional[MobileNetworkIds] = None allowedFixedNetworkIds: Optional[FixedNetworkIds] = None offeredAvailabilityZones: Optional[Annotated[list[ZoneDetails], Field(min_length=1)]] = None class FederationContextIdResponse(BaseModel): """``200`` body of ``GET /fed-context-id``. The spec's ``inline_response_200_2``. The wire field name is ``FederationContextId`` (capitalised) per the spec. """ FederationContextId: FederationContextId class FederationPatchRequest(BaseModel): """Body of ``PATCH /{federationContextId}/partner``. The spec's ``federationContextId_partner_body``. """ model_config = ConfigDict(extra="forbid") objectType: NetworkCodeObjectType operationType: NetworkCodeOperationType modificationDate: datetime addMobileNetworkIds: Optional[MobileNetworkIds] = None removeMobileNetworkIds: Optional[MobileNetworkIds] = None addFixedNetworkIds: Optional[FixedNetworkIds] = None removeFixedNetworkIds: Optional[FixedNetworkIds] = None
src/open_exposure_gateway/main.py +8 −0 Original line number Diff line number Diff line Loading @@ -44,6 +44,9 @@ from open_exposure_gateway.api.error_handlers import ( register_exception_handlers, x_correlator_header, ) from open_exposure_gateway.api.gsma.federation_manager.v1_2_0.router import ( router as federation_manager_router, ) from open_exposure_gateway.api.platform.health import router as health_router from open_exposure_gateway.application.services.edge_application_management_service import ( EdgeApplicationManagementService, Loading Loading @@ -159,6 +162,10 @@ openapi_tags = [ "name": "Location retrieval", "description": "Retrieve the location of a device", }, { "name": "Federation Manager", "description": "GSMA OPG Federation Management (EWBI) -- partner federation lifecycle", }, { "name": "Platform", "description": "Platform-specific endpoints (health, readiness probes)", Loading Loading @@ -309,6 +316,7 @@ def create_app(lifespan: Optional[Lifespan[FastAPI]] = None) -> FastAPI: app.include_router(edge_application_management_router) app.include_router(quality_on_demand_router) app.include_router(location_retrieval_router) app.include_router(federation_manager_router) return app Loading
tests/unit/test_federation_manager_endpoints.py 0 → 100644 +77 −0 Original line number Diff line number Diff line """GSMA Federation Manager -- surface stub. The FederationManagement routes are wired into the app and visible in the OpenAPI document, but every handler returns HTTP 501 until fulfilment logic is added. """ from datetime import datetime, timezone from typing import Any import pytest from fastapi.testclient import TestClient from pydantic import ValidationError from open_exposure_gateway.api.gsma.federation_manager.v1_2_0.router import BASE_PATH from open_exposure_gateway.api.gsma.federation_manager.v1_2_0.schemas import ( FederationRequestData, ) from open_exposure_gateway.main import app client = TestClient(app, raise_server_exceptions=False) _CTX = "fed-ctx-1" _VALID_CREATE = { "origOPFederationId": "orig-op-1", "initialDate": "2026-09-04T12:00:00Z", "partnerStatusLink": "https://orig-op.example.com/status", } _VALID_PATCH = { "objectType": "MOBILE_NETWORK_CODES", "operationType": "ADD_CODES", "modificationDate": "2026-09-04T12:00:00Z", } _ENDPOINTS = [ ("post", f"{BASE_PATH}/partner", _VALID_CREATE), ("get", f"{BASE_PATH}/{_CTX}/partner", None), ("patch", f"{BASE_PATH}/{_CTX}/partner", _VALID_PATCH), ("delete", f"{BASE_PATH}/{_CTX}/partner", None), ("get", f"{BASE_PATH}/fed-context-id", None), ] _OPERATION_IDS = { "create_federation", "get_federation_details", "update_federation", "delete_federation_details", "get_federation_context_id", } @pytest.mark.parametrize(("method", "path", "body"), _ENDPOINTS) def test_endpoint_returns_501(method: str, path: str, body: dict[str, Any] | None) -> None: response = client.request(method, path, json=body) assert response.status_code == 501 def test_openapi_exposes_federation_management_operations() -> None: paths = app.openapi()["paths"] seen = { op["operationId"] for methods in paths.values() for op in methods.values() if isinstance(op, dict) and "operationId" in op } assert _OPERATION_IDS <= seen def test_federation_request_data_round_trips() -> None: model = FederationRequestData.model_validate(_VALID_CREATE) assert model.origOPFederationId == "orig-op-1" assert model.initialDate == datetime(2026, 9, 4, 12, 0, tzinfo=timezone.utc) def test_federation_request_data_requires_partner_status_link() -> None: payload = {k: v for k, v in _VALID_CREATE.items() if k != "partnerStatusLink"} with pytest.raises(ValidationError): FederationRequestData.model_validate(payload)
