Loading src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/router.py +87 −4 Changes for src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/router.py: 87 added lines, 4 removed lines. Original line number Diff line number Diff line Loading @@ -22,14 +22,17 @@ response that passes is still relayed byte-for-byte. This module exposes the ``FederationManagement`` tag plus ``zone_subscribe``, ``get_zone_data``, ``get_zone_details`` and ``zone_unsubscribe`` from ``AvailabilityZoneInfoSynchronization``, plus two things that are not GSMA operations: ``get_federation_identity``, a read of our own identity, and the partner registry ``get_federation_identity``, a read of our own identity, the partner registry (``list_partners``, ``register_partner``, ``get_partner``, ``update_partner``), which the portal fills before it can federate with anyone. ``update_federation`` portal fills before it can federate with anyone, and the agreements that say what each partner may use here (``create_agreement``, ``update_agreement``; read embedded in ``get_partner``) with ``list_local_zones``, the zones an agreement can offer. ``update_federation`` (``UpdateFederation``) is out of scope. """ import json from typing import Annotated, Any from uuid import UUID import httpx import structlog Loading @@ -49,13 +52,16 @@ from open_exposure_gateway.api.gsma.federation_manager.v1_4_0.problems import ( problem_response, ) from open_exposure_gateway.api.gsma.federation_manager.v1_4_0.schemas import ( Agreement, FederationContextId, FederationContextIdResponse, FederationDetails, FederationIdentity, FederationRequestData, FederationResponseData, LocalZoneList, Partner, PartnerDetail, PartnerList, PartnerOpId, ProblemDetails, Loading Loading @@ -114,6 +120,12 @@ _FM_PARTNER_REGISTER_STATUSES = frozenset({201, 400, 409, 422, 500, 501}) _FM_PARTNER_GET_STATUSES = frozenset({200, 404, 500}) _FM_PARTNER_UPDATE_STATUSES = frozenset({200, 400, 404, 409, 422, 500, 501}) # Agreements are FM's own API too: 400 for a bad body, 404 for an unknown partner or # agreement, 422 for a validity period that ends before it starts. _FM_LOCAL_ZONES_STATUSES = frozenset({200, 500}) _FM_AGREEMENT_CREATE_STATUSES = frozenset({201, 400, 404, 422, 500}) _FM_AGREEMENT_UPDATE_STATUSES = frozenset({200, 400, 404, 422, 500}) _PROBLEM_JSON = "application/problem+json" Loading Loading @@ -310,7 +322,7 @@ async def register_partner( tags=["Federation Manager"], summary="Retrieve one registered partner operator platform", operation_id="get_partner", response_model=Partner, response_model=PartnerDetail, responses=_responses(400, 401, 404, 500, 502, 503), ) async def get_partner( Loading @@ -325,7 +337,7 @@ async def get_partner( f"/partners/{partnerOpId}", request, fm_statuses=_FM_PARTNER_GET_STATUSES, response_schema=Partner, response_schema=PartnerDetail, ) Loading Loading @@ -353,6 +365,77 @@ async def update_partner( ) @router.get( "/local-zones", tags=["Federation Manager"], summary="List our own zones, which an agreement can offer a partner", operation_id="list_local_zones", response_model=LocalZoneList, responses=_responses(401, 500, 502, 503), ) async def list_local_zones( request: Request, fm_client: FmClientDep, x_correlator: XCorrelatorHeader = None, ) -> Response: return await _relay( fm_client, "GET", "/local-zones", request, fm_statuses=_FM_LOCAL_ZONES_STATUSES, response_schema=LocalZoneList, ) @router.post( "/agreements", status_code=201, tags=["Federation Manager"], summary="Record what a partner may use here: zones, their GSMA terms, allowed requests", operation_id="create_agreement", response_model=Agreement, responses=_responses(400, 401, 404, 422, 500, 502, 503), ) async def create_agreement( request: Request, fm_client: FmClientDep, x_correlator: XCorrelatorHeader = None, ) -> Response: return await _relay( fm_client, "POST", "/agreements", request, fm_statuses=_FM_AGREEMENT_CREATE_STATUSES, response_schema=Agreement, ) @router.patch( "/agreements/{agreementId}", tags=["Federation Manager"], summary="Change an agreement's status, usage limits or validity", operation_id="update_agreement", response_model=Agreement, responses=_responses(400, 401, 404, 422, 500, 502, 503), ) async def update_agreement( agreementId: UUID, request: Request, fm_client: FmClientDep, x_correlator: XCorrelatorHeader = None, ) -> Response: return await _relay( fm_client, "PATCH", f"/agreements/{agreementId}", request, fm_statuses=_FM_AGREEMENT_UPDATE_STATUSES, response_schema=Agreement, ) @router.post( PARTNER_PREFIX, tags=["Federation Manager"], Loading src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/schemas.py +41 −1 Changes for src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/schemas.py: 41 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -16,7 +16,7 @@ from __future__ import annotations from datetime import datetime from enum import StrEnum from typing import Annotated, Literal, Optional from typing import Annotated, Any, Literal, Optional from uuid import UUID from pydantic import BaseModel, Field, RootModel, StringConstraints, model_validator Loading Loading @@ -255,6 +255,46 @@ class Partner(BaseModel): our_client_secret_set: bool = False class LocalZone(BaseModel): """One of our own zones, which an agreement can offer a partner.""" id: UUID name: str region: Optional[str] = None class LocalZoneList(RootModel[list[LocalZone]]): """``200`` body of ``GET /platform/v1/federation/local-zones``.""" AgreementStatus = Literal["draft", "active", "suspended", "expired"] class Agreement(BaseModel): """One federation agreement: what a partner may use at our platform, in FM's ``/internal`` field names. Like partner registrations, FM validates what is written and answers a bad body with a ``400`` ProblemDetails naming the field.""" id: UUID partner_op_id: UUID permitted_api_types: list[str] permitted_zone_ids: list[UUID] # per zone id: the GSMA terms offered there (reserved resources, quota, flavours) zone_resources: dict[str, dict[str, Any]] service_spec_mappings: dict[str, Any] usage_limits: dict[str, Any] valid_from: datetime valid_until: Optional[datetime] = None status: AgreementStatus class PartnerDetail(Partner): """``200`` body of ``GET /platform/v1/federation/partners/{partnerOpId}``: the partner plus every agreement with it, oldest first, whatever its status.""" agreements: list[Agreement] = [] class PartnerList(RootModel[list[Partner]]): """``200`` body of ``GET /platform/v1/federation/partners``.""" Loading src/open_exposure_gateway/core/authorization.py +3 −0 Changes for src/open_exposure_gateway/core/authorization.py: 3 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -108,6 +108,9 @@ OPERATION_REQUIREMENTS: dict[str, OperationRequirements] = { "POST /platform/v1/federation/partners": OperationRequirements("operator"), "GET /platform/v1/federation/partners/{partnerOpId}": OperationRequirements("operator"), "PATCH /platform/v1/federation/partners/{partnerOpId}": OperationRequirements("operator"), "GET /platform/v1/federation/local-zones": OperationRequirements("operator"), "POST /platform/v1/federation/agreements": OperationRequirements("operator"), "PATCH /platform/v1/federation/agreements/{agreementId}": OperationRequirements("operator"), "POST /platform/v1/federation/partners/{partnerOpId}/federations": ( OperationRequirements("operator") ), Loading tests/unit/test_federation_manager_endpoints.py +171 −0 Changes for tests/unit/test_federation_manager_endpoints.py: 171 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -182,6 +182,9 @@ _OPERATION_IDS = { "register_partner", "get_partner", "update_partner", "list_local_zones", "create_agreement", "update_agreement", "create_federation", "get_federation_context_id", "get_federation_details", Loading Loading @@ -902,3 +905,171 @@ def test_non_uuid_partner_op_id_on_the_registry_never_reaches_fm( assert response.status_code == 400 assert fake_fm_client.calls == [] _AGREEMENT = "0b8e5f7a-2c4d-4e6f-8a1b-3c5d7e9f1a2b" _AGREEMENT_BODY = ( b'{"id": "0b8e5f7a-2c4d-4e6f-8a1b-3c5d7e9f1a2b", ' b'"partner_op_id": "5f0c2a8e-3b1d-4c6e-9a7f-1d2e3f4a5b6c", ' b'"permitted_api_types": ["install-app"], ' b'"permitted_zone_ids": ["d3e8a7b7-9ffb-5b2e-b704-79e3ef61ca27"], ' b'"zone_resources": {"d3e8a7b7-9ffb-5b2e-b704-79e3ef61ca27": {"flavoursSupported": []}}, ' b'"service_spec_mappings": {"apps": [], "api_families": {}}, "usage_limits": {}, ' b'"valid_from": "2026-10-01T00:00:00Z", "valid_until": null, "status": "active"}' ) _AGREEMENT_ENDPOINTS = [ ("post", "/agreements", b'{"partner_op_id": "x"}', 201), ("patch", f"/agreements/{_AGREEMENT}", b'{"status": "suspended"}', 200), ] @pytest.mark.parametrize(("http_method", "path", "body", "status"), _AGREEMENT_ENDPOINTS) def test_agreements_relay_to_fm_unaltered( client: TestClient, fake_fm_client: _FakeFmClient, http_method: str, path: str, body: bytes, status: int, ) -> None: """FM owns the agreement rules, so OEG forwards any JSON and relays FM's answer.""" fake_fm_client.response = httpx.Response( status, content=_AGREEMENT_BODY, headers={"content-type": "application/json"} ) response = client.request( http_method, f"{BASE_PATH}{path}", content=body, headers={"content-type": "application/json"}, ) assert response.status_code == status assert response.content == _AGREEMENT_BODY call = fake_fm_client.calls[0] assert (call["method"], call["path"], call["content"]) == (http_method.upper(), path, body) @pytest.mark.parametrize(("http_method", "path", "body", "status"), _AGREEMENT_ENDPOINTS) @pytest.mark.parametrize("fm_status", [400, 404, 422, 500]) def test_agreement_errors_from_fm_are_relayed( client: TestClient, fake_fm_client: _FakeFmClient, http_method: str, path: str, body: bytes, status: int, fm_status: int, ) -> None: problem = b'{"type": "urn:oop:ewbi:error:x", "title": "X", "status": %d}' % fm_status fake_fm_client.response = httpx.Response( fm_status, content=problem, headers={"content-type": "application/problem+json"} ) response = client.request( http_method, f"{BASE_PATH}{path}", content=body, headers={"content-type": "application/json"}, ) assert response.status_code == fm_status assert response.content == problem @pytest.mark.parametrize(("http_method", "path", "body", "status"), _AGREEMENT_ENDPOINTS) @pytest.mark.parametrize( ("fm_status", "fm_body"), [(409, _AGREEMENT_BODY), (None, b"{}"), (None, _PARTNER_BODY)] ) def test_agreement_answers_outside_fm_contract_return_502( client: TestClient, fake_fm_client: _FakeFmClient, http_method: str, path: str, body: bytes, status: int, fm_status: int | None, fm_body: bytes, ) -> None: fake_fm_client.response = httpx.Response( fm_status or status, content=fm_body, headers={"content-type": "application/json"} ) _assert_bad_response_502( client.request( http_method, f"{BASE_PATH}{path}", content=body, headers={"content-type": "application/json"}, ) ) def test_non_uuid_agreement_id_never_reaches_fm( client: TestClient, fake_fm_client: _FakeFmClient ) -> None: response = client.patch( f"{BASE_PATH}/agreements/not-a-uuid", content=b'{"status": "active"}', headers={"content-type": "application/json"}, ) assert response.status_code == 400 assert fake_fm_client.calls == [] @pytest.mark.parametrize( "agreements", [b"", b', "agreements": []', b', "agreements": [' + _AGREEMENT_BODY + b"]"] ) def test_partner_detail_relays_embedded_agreements( client: TestClient, fake_fm_client: _FakeFmClient, agreements: bytes ) -> None: detail = _PARTNER_BODY[:-1] + agreements + b"}" fake_fm_client.response = httpx.Response( 200, content=detail, headers={"content-type": "application/json"} ) response = client.get(f"{BASE_PATH}/partners/{_PARTNER}") assert response.status_code == 200 assert response.content == detail def test_partner_detail_with_a_malformed_agreement_returns_502( client: TestClient, fake_fm_client: _FakeFmClient ) -> None: detail = _PARTNER_BODY[:-1] + b', "agreements": [{"id": "x"}]}' fake_fm_client.response = httpx.Response( 200, content=detail, headers={"content-type": "application/json"} ) _assert_bad_response_502(client.get(f"{BASE_PATH}/partners/{_PARTNER}")) _LOCAL_ZONES = b'[{"id": "d3e8a7b7-9ffb-5b2e-b704-79e3ef61ca27", "name": "Paris", "region": null}]' def test_local_zones_relay_to_fm_unaltered( client: TestClient, fake_fm_client: _FakeFmClient ) -> None: fake_fm_client.response = _fm_ok(_LOCAL_ZONES) response = client.get(f"{BASE_PATH}/local-zones") assert response.status_code == 200 assert response.content == _LOCAL_ZONES assert (fake_fm_client.calls[0]["method"], fake_fm_client.calls[0]["path"]) == ( "GET", "/local-zones", ) @pytest.mark.parametrize(("status", "body"), [(404, _LOCAL_ZONES), (200, b"{}")]) def test_local_zones_outside_fm_contract_return_502( client: TestClient, fake_fm_client: _FakeFmClient, status: int, body: bytes ) -> None: fake_fm_client.response = httpx.Response( status, content=body, headers={"content-type": "application/json"} ) _assert_bad_response_502(client.get(f"{BASE_PATH}/local-zones")) Loading
src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/router.py +87 −4 Changes for src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/router.py: 87 added lines, 4 removed lines. Original line number Diff line number Diff line Loading @@ -22,14 +22,17 @@ response that passes is still relayed byte-for-byte. This module exposes the ``FederationManagement`` tag plus ``zone_subscribe``, ``get_zone_data``, ``get_zone_details`` and ``zone_unsubscribe`` from ``AvailabilityZoneInfoSynchronization``, plus two things that are not GSMA operations: ``get_federation_identity``, a read of our own identity, and the partner registry ``get_federation_identity``, a read of our own identity, the partner registry (``list_partners``, ``register_partner``, ``get_partner``, ``update_partner``), which the portal fills before it can federate with anyone. ``update_federation`` portal fills before it can federate with anyone, and the agreements that say what each partner may use here (``create_agreement``, ``update_agreement``; read embedded in ``get_partner``) with ``list_local_zones``, the zones an agreement can offer. ``update_federation`` (``UpdateFederation``) is out of scope. """ import json from typing import Annotated, Any from uuid import UUID import httpx import structlog Loading @@ -49,13 +52,16 @@ from open_exposure_gateway.api.gsma.federation_manager.v1_4_0.problems import ( problem_response, ) from open_exposure_gateway.api.gsma.federation_manager.v1_4_0.schemas import ( Agreement, FederationContextId, FederationContextIdResponse, FederationDetails, FederationIdentity, FederationRequestData, FederationResponseData, LocalZoneList, Partner, PartnerDetail, PartnerList, PartnerOpId, ProblemDetails, Loading Loading @@ -114,6 +120,12 @@ _FM_PARTNER_REGISTER_STATUSES = frozenset({201, 400, 409, 422, 500, 501}) _FM_PARTNER_GET_STATUSES = frozenset({200, 404, 500}) _FM_PARTNER_UPDATE_STATUSES = frozenset({200, 400, 404, 409, 422, 500, 501}) # Agreements are FM's own API too: 400 for a bad body, 404 for an unknown partner or # agreement, 422 for a validity period that ends before it starts. _FM_LOCAL_ZONES_STATUSES = frozenset({200, 500}) _FM_AGREEMENT_CREATE_STATUSES = frozenset({201, 400, 404, 422, 500}) _FM_AGREEMENT_UPDATE_STATUSES = frozenset({200, 400, 404, 422, 500}) _PROBLEM_JSON = "application/problem+json" Loading Loading @@ -310,7 +322,7 @@ async def register_partner( tags=["Federation Manager"], summary="Retrieve one registered partner operator platform", operation_id="get_partner", response_model=Partner, response_model=PartnerDetail, responses=_responses(400, 401, 404, 500, 502, 503), ) async def get_partner( Loading @@ -325,7 +337,7 @@ async def get_partner( f"/partners/{partnerOpId}", request, fm_statuses=_FM_PARTNER_GET_STATUSES, response_schema=Partner, response_schema=PartnerDetail, ) Loading Loading @@ -353,6 +365,77 @@ async def update_partner( ) @router.get( "/local-zones", tags=["Federation Manager"], summary="List our own zones, which an agreement can offer a partner", operation_id="list_local_zones", response_model=LocalZoneList, responses=_responses(401, 500, 502, 503), ) async def list_local_zones( request: Request, fm_client: FmClientDep, x_correlator: XCorrelatorHeader = None, ) -> Response: return await _relay( fm_client, "GET", "/local-zones", request, fm_statuses=_FM_LOCAL_ZONES_STATUSES, response_schema=LocalZoneList, ) @router.post( "/agreements", status_code=201, tags=["Federation Manager"], summary="Record what a partner may use here: zones, their GSMA terms, allowed requests", operation_id="create_agreement", response_model=Agreement, responses=_responses(400, 401, 404, 422, 500, 502, 503), ) async def create_agreement( request: Request, fm_client: FmClientDep, x_correlator: XCorrelatorHeader = None, ) -> Response: return await _relay( fm_client, "POST", "/agreements", request, fm_statuses=_FM_AGREEMENT_CREATE_STATUSES, response_schema=Agreement, ) @router.patch( "/agreements/{agreementId}", tags=["Federation Manager"], summary="Change an agreement's status, usage limits or validity", operation_id="update_agreement", response_model=Agreement, responses=_responses(400, 401, 404, 422, 500, 502, 503), ) async def update_agreement( agreementId: UUID, request: Request, fm_client: FmClientDep, x_correlator: XCorrelatorHeader = None, ) -> Response: return await _relay( fm_client, "PATCH", f"/agreements/{agreementId}", request, fm_statuses=_FM_AGREEMENT_UPDATE_STATUSES, response_schema=Agreement, ) @router.post( PARTNER_PREFIX, tags=["Federation Manager"], Loading
src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/schemas.py +41 −1 Changes for src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/schemas.py: 41 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -16,7 +16,7 @@ from __future__ import annotations from datetime import datetime from enum import StrEnum from typing import Annotated, Literal, Optional from typing import Annotated, Any, Literal, Optional from uuid import UUID from pydantic import BaseModel, Field, RootModel, StringConstraints, model_validator Loading Loading @@ -255,6 +255,46 @@ class Partner(BaseModel): our_client_secret_set: bool = False class LocalZone(BaseModel): """One of our own zones, which an agreement can offer a partner.""" id: UUID name: str region: Optional[str] = None class LocalZoneList(RootModel[list[LocalZone]]): """``200`` body of ``GET /platform/v1/federation/local-zones``.""" AgreementStatus = Literal["draft", "active", "suspended", "expired"] class Agreement(BaseModel): """One federation agreement: what a partner may use at our platform, in FM's ``/internal`` field names. Like partner registrations, FM validates what is written and answers a bad body with a ``400`` ProblemDetails naming the field.""" id: UUID partner_op_id: UUID permitted_api_types: list[str] permitted_zone_ids: list[UUID] # per zone id: the GSMA terms offered there (reserved resources, quota, flavours) zone_resources: dict[str, dict[str, Any]] service_spec_mappings: dict[str, Any] usage_limits: dict[str, Any] valid_from: datetime valid_until: Optional[datetime] = None status: AgreementStatus class PartnerDetail(Partner): """``200`` body of ``GET /platform/v1/federation/partners/{partnerOpId}``: the partner plus every agreement with it, oldest first, whatever its status.""" agreements: list[Agreement] = [] class PartnerList(RootModel[list[Partner]]): """``200`` body of ``GET /platform/v1/federation/partners``.""" Loading
src/open_exposure_gateway/core/authorization.py +3 −0 Changes for src/open_exposure_gateway/core/authorization.py: 3 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -108,6 +108,9 @@ OPERATION_REQUIREMENTS: dict[str, OperationRequirements] = { "POST /platform/v1/federation/partners": OperationRequirements("operator"), "GET /platform/v1/federation/partners/{partnerOpId}": OperationRequirements("operator"), "PATCH /platform/v1/federation/partners/{partnerOpId}": OperationRequirements("operator"), "GET /platform/v1/federation/local-zones": OperationRequirements("operator"), "POST /platform/v1/federation/agreements": OperationRequirements("operator"), "PATCH /platform/v1/federation/agreements/{agreementId}": OperationRequirements("operator"), "POST /platform/v1/federation/partners/{partnerOpId}/federations": ( OperationRequirements("operator") ), Loading
tests/unit/test_federation_manager_endpoints.py +171 −0 Changes for tests/unit/test_federation_manager_endpoints.py: 171 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -182,6 +182,9 @@ _OPERATION_IDS = { "register_partner", "get_partner", "update_partner", "list_local_zones", "create_agreement", "update_agreement", "create_federation", "get_federation_context_id", "get_federation_details", Loading Loading @@ -902,3 +905,171 @@ def test_non_uuid_partner_op_id_on_the_registry_never_reaches_fm( assert response.status_code == 400 assert fake_fm_client.calls == [] _AGREEMENT = "0b8e5f7a-2c4d-4e6f-8a1b-3c5d7e9f1a2b" _AGREEMENT_BODY = ( b'{"id": "0b8e5f7a-2c4d-4e6f-8a1b-3c5d7e9f1a2b", ' b'"partner_op_id": "5f0c2a8e-3b1d-4c6e-9a7f-1d2e3f4a5b6c", ' b'"permitted_api_types": ["install-app"], ' b'"permitted_zone_ids": ["d3e8a7b7-9ffb-5b2e-b704-79e3ef61ca27"], ' b'"zone_resources": {"d3e8a7b7-9ffb-5b2e-b704-79e3ef61ca27": {"flavoursSupported": []}}, ' b'"service_spec_mappings": {"apps": [], "api_families": {}}, "usage_limits": {}, ' b'"valid_from": "2026-10-01T00:00:00Z", "valid_until": null, "status": "active"}' ) _AGREEMENT_ENDPOINTS = [ ("post", "/agreements", b'{"partner_op_id": "x"}', 201), ("patch", f"/agreements/{_AGREEMENT}", b'{"status": "suspended"}', 200), ] @pytest.mark.parametrize(("http_method", "path", "body", "status"), _AGREEMENT_ENDPOINTS) def test_agreements_relay_to_fm_unaltered( client: TestClient, fake_fm_client: _FakeFmClient, http_method: str, path: str, body: bytes, status: int, ) -> None: """FM owns the agreement rules, so OEG forwards any JSON and relays FM's answer.""" fake_fm_client.response = httpx.Response( status, content=_AGREEMENT_BODY, headers={"content-type": "application/json"} ) response = client.request( http_method, f"{BASE_PATH}{path}", content=body, headers={"content-type": "application/json"}, ) assert response.status_code == status assert response.content == _AGREEMENT_BODY call = fake_fm_client.calls[0] assert (call["method"], call["path"], call["content"]) == (http_method.upper(), path, body) @pytest.mark.parametrize(("http_method", "path", "body", "status"), _AGREEMENT_ENDPOINTS) @pytest.mark.parametrize("fm_status", [400, 404, 422, 500]) def test_agreement_errors_from_fm_are_relayed( client: TestClient, fake_fm_client: _FakeFmClient, http_method: str, path: str, body: bytes, status: int, fm_status: int, ) -> None: problem = b'{"type": "urn:oop:ewbi:error:x", "title": "X", "status": %d}' % fm_status fake_fm_client.response = httpx.Response( fm_status, content=problem, headers={"content-type": "application/problem+json"} ) response = client.request( http_method, f"{BASE_PATH}{path}", content=body, headers={"content-type": "application/json"}, ) assert response.status_code == fm_status assert response.content == problem @pytest.mark.parametrize(("http_method", "path", "body", "status"), _AGREEMENT_ENDPOINTS) @pytest.mark.parametrize( ("fm_status", "fm_body"), [(409, _AGREEMENT_BODY), (None, b"{}"), (None, _PARTNER_BODY)] ) def test_agreement_answers_outside_fm_contract_return_502( client: TestClient, fake_fm_client: _FakeFmClient, http_method: str, path: str, body: bytes, status: int, fm_status: int | None, fm_body: bytes, ) -> None: fake_fm_client.response = httpx.Response( fm_status or status, content=fm_body, headers={"content-type": "application/json"} ) _assert_bad_response_502( client.request( http_method, f"{BASE_PATH}{path}", content=body, headers={"content-type": "application/json"}, ) ) def test_non_uuid_agreement_id_never_reaches_fm( client: TestClient, fake_fm_client: _FakeFmClient ) -> None: response = client.patch( f"{BASE_PATH}/agreements/not-a-uuid", content=b'{"status": "active"}', headers={"content-type": "application/json"}, ) assert response.status_code == 400 assert fake_fm_client.calls == [] @pytest.mark.parametrize( "agreements", [b"", b', "agreements": []', b', "agreements": [' + _AGREEMENT_BODY + b"]"] ) def test_partner_detail_relays_embedded_agreements( client: TestClient, fake_fm_client: _FakeFmClient, agreements: bytes ) -> None: detail = _PARTNER_BODY[:-1] + agreements + b"}" fake_fm_client.response = httpx.Response( 200, content=detail, headers={"content-type": "application/json"} ) response = client.get(f"{BASE_PATH}/partners/{_PARTNER}") assert response.status_code == 200 assert response.content == detail def test_partner_detail_with_a_malformed_agreement_returns_502( client: TestClient, fake_fm_client: _FakeFmClient ) -> None: detail = _PARTNER_BODY[:-1] + b', "agreements": [{"id": "x"}]}' fake_fm_client.response = httpx.Response( 200, content=detail, headers={"content-type": "application/json"} ) _assert_bad_response_502(client.get(f"{BASE_PATH}/partners/{_PARTNER}")) _LOCAL_ZONES = b'[{"id": "d3e8a7b7-9ffb-5b2e-b704-79e3ef61ca27", "name": "Paris", "region": null}]' def test_local_zones_relay_to_fm_unaltered( client: TestClient, fake_fm_client: _FakeFmClient ) -> None: fake_fm_client.response = _fm_ok(_LOCAL_ZONES) response = client.get(f"{BASE_PATH}/local-zones") assert response.status_code == 200 assert response.content == _LOCAL_ZONES assert (fake_fm_client.calls[0]["method"], fake_fm_client.calls[0]["path"]) == ( "GET", "/local-zones", ) @pytest.mark.parametrize(("status", "body"), [(404, _LOCAL_ZONES), (200, b"{}")]) def test_local_zones_outside_fm_contract_return_502( client: TestClient, fake_fm_client: _FakeFmClient, status: int, body: bytes ) -> None: fake_fm_client.response = httpx.Response( status, content=body, headers={"content-type": "application/json"} ) _assert_bad_response_502(client.get(f"{BASE_PATH}/local-zones"))
