feat: add get_zone_data and zone_unsubscribe stubs (ba260cc4) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/api/gsma/federation_manager/v1_2_0/router.py

+38 −4

Original line number	Diff line number	Diff line
		@@ -3,10 +3,11 @@
		Endpoints defined by ``API_definitions/federation-manager.yaml``.

		This module currently exposes the ``FederationManagement`` tag plus
		``zone_subscribe`` from ``AvailabilityZoneInfoSynchronization``, 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.
		``zone_subscribe``, ``get_zone_data`` and ``zone_unsubscribe`` from
		``AvailabilityZoneInfoSynchronization``, 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
		@@ -20,6 +21,8 @@ from open_exposure_gateway.api.gsma.federation_manager.v1_2_0.schemas import (
		FederationRequestData,
		FederationResponseData,
		ProblemDetails,
		ZoneIdentifier,
		ZoneRegisteredData,
		ZoneRegistrationRequestData,
		ZoneRegistrationResponseData,
		)
		@@ -117,3 +120,34 @@ async def zone_subscribe(
		federationContextId: FederationContextId, request: ZoneRegistrationRequestData
		) -> Any:
		raise NotImplementedException(message=_NOT_IMPLEMENTED)


		@router.get(
		"/{federationContextId}/zones/{zoneId}",
		tags=["Availability Zone Info Synchronization"],
		summary=(
		"Retrieves details about the computation and network resources that partner "
		"OP has reserved for this zone"
		),
		operation_id="get_zone_data",
		response_model=ZoneRegisteredData,
		response_model_exclude_none=True,
		responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520),
		)
		async def get_zone_data(federationContextId: FederationContextId, zoneId: ZoneIdentifier) -> Any:
		raise NotImplementedException(message=_NOT_IMPLEMENTED)


		@router.delete(
		"/{federationContextId}/zones/{zoneId}",
		tags=["Availability Zone Info Synchronization"],
		summary=(
		"Assert usage of a partner OP zone. Originating OP informs partner OP that "
		"it will no longer access the specified zone"
		),
		operation_id="zone_unsubscribe",
		status_code=200,
		responses=_responses(400, 401, 404, 409, 422, 500, 501, 503, 520),
		)
		async def zone_unsubscribe(federationContextId: FederationContextId, zoneId: ZoneIdentifier) -> Any:
		raise NotImplementedException(message=_NOT_IMPLEMENTED)

tests/unit/test_federation_manager_endpoints.py

+5 −0

Original line number	Diff line number	Diff line
		@@ -21,6 +21,7 @@ from open_exposure_gateway.main import app
		client = TestClient(app, raise_server_exceptions=False)

		_CTX = "fed-ctx-1"
		_ZONE = "zone-a"
		_VALID_CREATE = {
		"origOPFederationId": "orig-op-1",
		"initialDate": "2026-09-04T12:00:00Z",
		@@ -37,6 +38,8 @@ _ENDPOINTS = [
		("delete", f"{BASE_PATH}/{_CTX}/partner", None),
		("get", f"{BASE_PATH}/fed-context-id", None),
		("post", f"{BASE_PATH}/{_CTX}/zones", _VALID_ZONE_SUBSCRIBE),
		("get", f"{BASE_PATH}/{_CTX}/zones/{_ZONE}", None),
		("delete", f"{BASE_PATH}/{_CTX}/zones/{_ZONE}", None),
		]

		_OPERATION_IDS = {
		@@ -45,6 +48,8 @@ _OPERATION_IDS = {
		"delete_federation_details",
		"get_federation_context_id",
		"zone_subscribe",
		"get_zone_data",
		"zone_unsubscribe",
		}