feat(fm): answer GetFederationDetails as the partner (507f6d58) · Commits · OOP / code / federation-manager

src/federation_manager/api/ewbi/v1/management.py

+19 −0

Original line number	Diff line number	Diff line
		@@ -7,6 +7,7 @@ from federation_manager.api.security import get_bearer_token
		from federation_manager.application.authentication import PartnerAuthenticator
		from federation_manager.application.federation import InboundFederationService
		from federation_manager.contracts.ewbi import (
		FederationDetails,
		FederationHealthInfo,
		FederationHealthResponse,
		FederationRequestData,
		@@ -74,6 +75,24 @@ async def create_federation(
		return accepted


		@router.get(
		"/{federationContextId}/partner",
		operation_id="GetFederationDetails",
		response_model=FederationDetails,
		# every field is optional in OPG.04 and none is nullable: absent means left out
		response_model_exclude_none=True,
		responses=EWBI_ERROR_RESPONSES,
		)
		async def get_federation_details(
		federationContextId: FederationContextIdPath, # noqa: N803 - GSMA path template name
		auth: Annotated[PartnerAuthenticator, Depends(get_partner_authenticator)],
		service: Annotated[InboundFederationService, Depends(get_inbound_federation_service)],
		token: Annotated[str, Depends(get_bearer_token)],
		) -> FederationDetails:
		partner = await auth.authenticate(token)
		return await service.details(partner, federationContextId)


		@router.delete(
		"/{federationContextId}/partner",
		operation_id="DeleteFederationDetails",

src/federation_manager/application/federation.py

+19 −1

Original line number	Diff line number	Diff line
		@@ -6,6 +6,7 @@ from uuid import UUID, uuid4
		from pydantic import ValidationError

		from federation_manager.contracts.ewbi import (
		FederationDetails,
		FederationRequestData,
		FederationResponseData,
		MobileNetworkIds,
		@@ -216,8 +217,25 @@ class InboundFederationService:
		partner_op_federation_id=self._local.federation_id,
		)

		async def details(self, partner: PartnerOP, federation_context_id: str) -> FederationDetails:
		await self._live_inbound(partner, federation_context_id)
		# OPG.04 asks for offered zones and our edge-discovery/LCM endpoints too; FM has no
		# zone catalogue or endpoint configuration yet, so those stay absent rather than empty.
		return FederationDetails(
		platform_caps=list(self._local.platform_caps),
		allowed_mobile_network_ids=MobileNetworkIds(
		mcc=self._local.mcc, mncs=list(self._local.mncs)
		),
		)

		async def terminate(self, partner: PartnerOP, federation_context_id: str) -> None:
		context = await self._live_inbound(partner, federation_context_id)
		await self._contexts.set_status(context.id, TERMINATED)

		async def _live_inbound(
		self, partner: PartnerOP, federation_context_id: str
		) -> FederationContext:
		context = await self._contexts.find_inbound(partner.id, federation_context_id)
		if context is None or context.is_terminated():
		raise FederationContextUnknown(partner.id)
		await self._contexts.set_status(context.id, TERMINATED)
		return context

src/federation_manager/contracts/ewbi.py

+5 −0

Original line number	Diff line number	Diff line
		@@ -182,6 +182,11 @@ class FederationDetails(BaseModel):
		platform_caps: list[str] \| None = Field(
		default=None, validation_alias="platformCaps", serialization_alias="platformCaps"
		)
		allowed_mobile_network_ids: MobileNetworkIds \| None = Field(
		default=None,
		validation_alias="allowedMobileNetworkIds",
		serialization_alias="allowedMobileNetworkIds",
		)


		class ZoneRegistrationRequestData(BaseModel):

tests/test_inbound_federation_details.py

0 → 100644

+106 −0

Original line number	Diff line number	Diff line
		from collections.abc import AsyncIterator
		from contextlib import asynccontextmanager
		from datetime import datetime, timezone
		from uuid import uuid4

		from fastapi import FastAPI
		from fastapi.testclient import TestClient

		from federation_manager.application.federation import LocalOperator
		from federation_manager.dependencies import (
		get_federation_context_repo,
		get_jwt_validator,
		get_local_operator,
		get_partner_repo,
		)
		from federation_manager.domain.models import FederationContext, PartnerOP, ValidatedClaims
		from federation_manager.main import create_app
		from tests.fakes import FakeJwtValidator, InMemoryFederationContextRepo, InMemoryPartnerRepo

		CLIENT_ID = "partner-a"
		OTHER_CLIENT_ID = "partner-b"
		TOKEN = "token-partner-a"
		OTHER_TOKEN = "token-partner-b"
		CONTEXT_ID = "fed-ctx-1"
		URL = f"/operatorplatform/federation/v1/{CONTEXT_ID}/partner"
		LOCAL = LocalOperator(
		federation_id="oop-i2cat",
		country_code="ES",
		mcc="214",
		mncs=("07", "01"),
		partner_status_link="https://us.example/operatorplatform/federation/v1/partner-status",
		)


		@asynccontextmanager
		async def _no_infra(app: FastAPI) -> AsyncIterator[None]:
		yield


		def _bearer(token: str) -> dict[str, str]:
		return {"Authorization": f"Bearer {token}"}


		def _client(context_status: str = "available", direction: str = "inbound") -> TestClient:
		partner = PartnerOP(id=uuid4(), mcc_mnc="208-01", oauth2_client_id=CLIENT_ID, status="active")
		other = PartnerOP(
		id=uuid4(), mcc_mnc="234-15", oauth2_client_id=OTHER_CLIENT_ID, status="active"
		)
		context = FederationContext(
		id=uuid4(),
		partner_op_id=partner.id,
		direction=direction,
		federation_context_id=CONTEXT_ID,
		status=context_status,
		created_at=datetime(2026, 3, 1, tzinfo=timezone.utc),
		)
		validator = FakeJwtValidator(
		{
		TOKEN: ValidatedClaims(client_id=CLIENT_ID, scopes={"fed-mgmt"}),
		OTHER_TOKEN: ValidatedClaims(client_id=OTHER_CLIENT_ID, scopes={"fed-mgmt"}),
		}
		)
		app = create_app(lifespan=_no_infra)
		app.dependency_overrides[get_partner_repo] = lambda: InMemoryPartnerRepo([partner, other])
		app.dependency_overrides[get_jwt_validator] = lambda: validator
		app.dependency_overrides[get_local_operator] = lambda: LOCAL
		app.dependency_overrides[get_federation_context_repo] = lambda: InMemoryFederationContextRepo(
		[context]
		)
		return TestClient(app)


		def test_details_describe_this_operator_to_the_partner() -> None:
		r = _client().get(URL, headers=_bearer(TOKEN))

		assert r.status_code == 200
		# no zone catalogue or endpoint config yet, so those keys are absent, never null
		assert r.json() == {
		"platformCaps": ["serviceAPIs"],
		"allowedMobileNetworkIds": {"mcc": "214", "mncs": ["07", "01"]},
		}


		def test_terminated_context_is_404() -> None:
		r = _client(context_status="terminated").get(URL, headers=_bearer(TOKEN))

		assert r.status_code == 404
		assert r.json()["type"] == "urn:oop:ewbi:error:federation-context-unknown"


		def test_our_outbound_context_is_not_readable_by_the_partner() -> None:
		# the same id on our outbound side is our view of their federation, not one they created
		r = _client(direction="outbound").get(URL, headers=_bearer(TOKEN))

		assert r.status_code == 404


		def test_another_partners_context_is_not_visible() -> None:
		r = _client().get(URL, headers=_bearer(OTHER_TOKEN))

		assert r.status_code == 404
		assert "platformCaps" not in r.text


		def test_missing_token_is_401() -> None:
		assert _client().get(URL).status_code == 401