Loading tests/integration/test_two_stack_federation.py +66 −3 Original line number Diff line number Diff line Loading @@ -112,21 +112,23 @@ def _fm_env(database: str, port: int, federation_id: str, **extra: str) -> dict[ "FM_COUNTRY_CODE": "ES", "FM_MCC": "214", "FM_MNCS": '["07"]', "FM_PARTNER_STATUS_LINK": f"http://127.0.0.1:{port}/operatorplatform/federation/v1/partner-status", "FM_PARTNER_STATUS_LINK": ( f"http://127.0.0.1:{port}/operatorplatform/federation/v1/callbacks/partner-status" ), "FM_ALLOW_INSECURE_PARTNER_ENDPOINTS": "true", "FM_EVENT_CONSUMER_DURABLE": f"fm-event-worker-{federation_id}", **extra, } def _access_token(client_id: str, secret: str) -> str: def _access_token(client_id: str, secret: str, scope: str = "fed-mgmt") -> str: response = httpx.post( TOKEN_ENDPOINT, data={ "grant_type": "client_credentials", "client_id": client_id, "client_secret": secret, "scope": "fed-mgmt", "scope": scope, }, timeout=10.0, ) Loading Loading @@ -272,3 +274,64 @@ def test_partner_b_rejects_an_unknown_client(stacks: Stacks) -> None: assert health.status_code in (401, 404) assert "federationHealthStatus" not in health.text async def _outbound(database: str, partner_id: UUID, federation_context_id: str) -> str | None: """Status of one outbound context, whatever it is: _context() only sees available ones.""" engine = build_engine(f"{PG_ROOT}/{database}") async with build_session_maker(engine)() as session: context = await PostgresFederationContextRepo(session).find_outbound( partner_id, federation_context_id ) await engine.dispose() return None if context is None else context.status def _partner_status_event(context_id: str, status: str) -> dict[str, str]: return { "federationContextId": context_id, "objectType": "FEDERATION", "operationType": "STATUS", "federationStatus": status, "modificationDate": "2026-09-23T09:00:00Z", } def test_partner_reports_a_status_change_over_the_callback(stacks: Stacks) -> None: outbound = asyncio.run(_context(DB_A, stacks.partner_b, "outbound")) assert outbound is not None, "bootstrap federation failed, nothing to report on" url = ( f"http://127.0.0.1:{stacks.port_a}/operatorplatform/federation/v1/callbacks/partner-status" ) event = _partner_status_event(outbound.federation_context_id, "LOCKED") accepted = httpx.post( url, json=event, headers={"Authorization": f"Bearer {_access_token(CLIENT_B, SECRET_B, 'fed-mgmt-notif')}"}, timeout=10.0, ) assert accepted.status_code == 204, accepted.text assert ( asyncio.run(_outbound(DB_A, stacks.partner_b, outbound.federation_context_id)) == "locked" ) def test_callback_needs_the_notification_scope(stacks: Stacks) -> None: outbound = asyncio.run(_context(DB_A, stacks.partner_b, "outbound")) assert outbound is not None url = ( f"http://127.0.0.1:{stacks.port_a}/operatorplatform/federation/v1/callbacks/partner-status" ) rejected = httpx.post( url, json=_partner_status_event(outbound.federation_context_id, "FAILED"), headers={"Authorization": f"Bearer {_access_token(CLIENT_B, SECRET_B)}"}, timeout=10.0, ) assert rejected.status_code == 401 status = asyncio.run(_outbound(DB_A, stacks.partner_b, outbound.federation_context_id)) assert status == "available" Loading
tests/integration/test_two_stack_federation.py +66 −3 Original line number Diff line number Diff line Loading @@ -112,21 +112,23 @@ def _fm_env(database: str, port: int, federation_id: str, **extra: str) -> dict[ "FM_COUNTRY_CODE": "ES", "FM_MCC": "214", "FM_MNCS": '["07"]', "FM_PARTNER_STATUS_LINK": f"http://127.0.0.1:{port}/operatorplatform/federation/v1/partner-status", "FM_PARTNER_STATUS_LINK": ( f"http://127.0.0.1:{port}/operatorplatform/federation/v1/callbacks/partner-status" ), "FM_ALLOW_INSECURE_PARTNER_ENDPOINTS": "true", "FM_EVENT_CONSUMER_DURABLE": f"fm-event-worker-{federation_id}", **extra, } def _access_token(client_id: str, secret: str) -> str: def _access_token(client_id: str, secret: str, scope: str = "fed-mgmt") -> str: response = httpx.post( TOKEN_ENDPOINT, data={ "grant_type": "client_credentials", "client_id": client_id, "client_secret": secret, "scope": "fed-mgmt", "scope": scope, }, timeout=10.0, ) Loading Loading @@ -272,3 +274,64 @@ def test_partner_b_rejects_an_unknown_client(stacks: Stacks) -> None: assert health.status_code in (401, 404) assert "federationHealthStatus" not in health.text async def _outbound(database: str, partner_id: UUID, federation_context_id: str) -> str | None: """Status of one outbound context, whatever it is: _context() only sees available ones.""" engine = build_engine(f"{PG_ROOT}/{database}") async with build_session_maker(engine)() as session: context = await PostgresFederationContextRepo(session).find_outbound( partner_id, federation_context_id ) await engine.dispose() return None if context is None else context.status def _partner_status_event(context_id: str, status: str) -> dict[str, str]: return { "federationContextId": context_id, "objectType": "FEDERATION", "operationType": "STATUS", "federationStatus": status, "modificationDate": "2026-09-23T09:00:00Z", } def test_partner_reports_a_status_change_over_the_callback(stacks: Stacks) -> None: outbound = asyncio.run(_context(DB_A, stacks.partner_b, "outbound")) assert outbound is not None, "bootstrap federation failed, nothing to report on" url = ( f"http://127.0.0.1:{stacks.port_a}/operatorplatform/federation/v1/callbacks/partner-status" ) event = _partner_status_event(outbound.federation_context_id, "LOCKED") accepted = httpx.post( url, json=event, headers={"Authorization": f"Bearer {_access_token(CLIENT_B, SECRET_B, 'fed-mgmt-notif')}"}, timeout=10.0, ) assert accepted.status_code == 204, accepted.text assert ( asyncio.run(_outbound(DB_A, stacks.partner_b, outbound.federation_context_id)) == "locked" ) def test_callback_needs_the_notification_scope(stacks: Stacks) -> None: outbound = asyncio.run(_context(DB_A, stacks.partner_b, "outbound")) assert outbound is not None url = ( f"http://127.0.0.1:{stacks.port_a}/operatorplatform/federation/v1/callbacks/partner-status" ) rejected = httpx.post( url, json=_partner_status_event(outbound.federation_context_id, "FAILED"), headers={"Authorization": f"Bearer {_access_token(CLIENT_B, SECRET_B)}"}, timeout=10.0, ) assert rejected.status_code == 401 status = asyncio.run(_outbound(DB_A, stacks.partner_b, outbound.federation_context_id)) assert status == "available"
