test(fm): one auth guarantee across all EWBI routes (3a7e32ce) · Commits · OOP / code / federation-manager

tests/test_create_federation.py

+0 −39

Original line number	Diff line number	Diff line
		@@ -130,33 +130,6 @@ def test_a_terminated_federation_can_be_re_established() -> None:
		assert len(contexts.contexts) == 2


		def test_unknown_partner_is_401() -> None:
		client, _, contexts = _client()

		r = client.post(URL, json=_body(), headers=_bearer(STRANGER_TOKEN))

		assert r.status_code == 401
		assert r.json()["type"] == "urn:oop:ewbi:error:partner-unknown"


		def test_missing_token_is_401() -> None:
		client, _, _ = _client()

		r = client.post(URL, json=_body())

		assert r.status_code == 401
		assert r.headers["WWW-Authenticate"] == 'Bearer scope="fed-mgmt"'


		def test_suspended_partner_is_403() -> None:
		client, _, _ = _client(partner_status="suspended")

		r = client.post(URL, json=_body(), headers=_bearer(TOKEN))

		assert r.status_code == 403
		assert r.json()["type"] == "urn:oop:ewbi:error:partner-not-active"


		def test_request_without_mandatory_fields_is_422() -> None:
		client, _, _ = _client()

		@@ -232,18 +205,6 @@ def test_another_partner_cannot_delete_our_federation() -> None:
		assert contexts.contexts[0].status == "available"


		def test_delete_without_a_token_is_401() -> None:
		client, _, contexts = _client()
		context_id = client.post(URL, json=_body(), headers=_bearer(TOKEN)).json()[
		"federationContextId"
		]

		r = client.delete(_delete_url(context_id))

		assert r.status_code == 401
		assert contexts.contexts[0].status == "available"


		def test_health_still_reports_a_locked_federation() -> None:
		client, partner, contexts = _client()
		client.post(URL, json=_body(), headers=_bearer(TOKEN))

tests/test_ewbi_auth.py

0 → 100644

+176 −0

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

		import pytest
		from fastapi import FastAPI
		from fastapi.testclient import TestClient

		from federation_manager.dependencies import (
		get_agreement_repo,
		get_command_publisher,
		get_federation_context_repo,
		get_jwt_validator,
		get_partner_repo,
		get_transaction_repo,
		)
		from federation_manager.domain.models import (
		Agreement,
		AppMapping,
		FederationContext,
		PartnerOP,
		ValidatedClaims,
		)
		from federation_manager.main import create_app
		from tests.fakes import (
		FakeJwtValidator,
		InMemoryAgreementRepo,
		InMemoryFederationContextRepo,
		InMemoryPartnerRepo,
		InMemoryTransactionRepo,
		RecordingCommandPublisher,
		)

		CLIENT_ID = "partner-a"
		TOKEN = "token-partner-a"
		STRANGER_TOKEN = "token-partner-nobody"
		CONTEXT_ID = "fed-ctx-1"
		BASE = "/operatorplatform/federation/v1"
		ZONE_ID = str(uuid4())
		SPEC_ID = uuid4()

		_CREATE_FEDERATION_BODY = {
		"initialDate": "2026-09-08T12:00:00Z",
		"partnerStatusLink": "https://partner.example/partner-status",
		}
		_INSTALL_APP_BODY = {
		"appId": "videoAnalytics",
		"appVersion": "1.2.0",
		"appProviderId": "partnerProvider",
		"zoneInfo": {"zoneId": ZONE_ID, "flavourId": "small"},
		"appInstCallbackLink": "https://partner.example/cb",
		}

		# Every partner-facing EWBI route, with a body valid enough that authentication is what fails.
		ROUTES = [
		("GET", f"{BASE}/{CONTEXT_ID}/health", None, {}),
		("POST", f"{BASE}/partner", _CREATE_FEDERATION_BODY, {}),
		("DELETE", f"{BASE}/{CONTEXT_ID}/partner", None, {}),
		(
		"POST",
		f"{BASE}/{CONTEXT_ID}/application/lcm",
		_INSTALL_APP_BODY,
		{"Idempotency-Key": "idem-1"},
		),
		]
		ROUTE_IDS = [f"{method} {path}" for method, path, _, _ in ROUTES]


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


		def _client(partner_status: str = "active") -> TestClient:
		partner = PartnerOP(
		id=uuid4(), mcc_mnc="214-07", oauth2_client_id=CLIENT_ID, status=partner_status
		)
		context = FederationContext(
		id=uuid4(),
		partner_op_id=partner.id,
		direction="inbound",
		federation_context_id=CONTEXT_ID,
		status="available",
		created_at=datetime(2026, 3, 1, tzinfo=timezone.utc),
		)
		agreement = Agreement(
		id=uuid4(),
		partner_op_id=partner.id,
		permitted_api_types={"install-app"},
		permitted_zone_ids={UUID(ZONE_ID)},
		app_mappings=(
		AppMapping(
		app_id="videoAnalytics",
		app_version="1.2.0",
		flavour_id="small",
		service_specification_id=SPEC_ID,
		),
		),
		api_family_mappings={},
		valid_from=datetime(2026, 1, 1, tzinfo=timezone.utc),
		valid_until=None,
		status="active",
		)
		validator = FakeJwtValidator(
		{
		TOKEN: ValidatedClaims(client_id=CLIENT_ID, scopes={"fed-mgmt"}),
		STRANGER_TOKEN: ValidatedClaims(client_id="partner-nobody", scopes={"fed-mgmt"}),
		}
		)

		app = create_app(lifespan=_no_infra)
		app.dependency_overrides[get_partner_repo] = lambda: InMemoryPartnerRepo([partner])
		app.dependency_overrides[get_jwt_validator] = lambda: validator
		app.dependency_overrides[get_federation_context_repo] = lambda: InMemoryFederationContextRepo(
		[context]
		)
		app.dependency_overrides[get_agreement_repo] = lambda: InMemoryAgreementRepo([agreement])
		app.dependency_overrides[get_transaction_repo] = lambda: InMemoryTransactionRepo()
		app.dependency_overrides[get_command_publisher] = lambda: RecordingCommandPublisher()
		return TestClient(app)


		def _call(
		client: TestClient,
		route: tuple[str, str, dict[str, Any] \| None, dict[str, str]],
		token: str \| None,
		) -> Any:
		method, path, body, headers = route
		request_headers = dict(headers)
		if token is not None:
		request_headers["Authorization"] = f"Bearer {token}"
		return client.request(method, path, json=body, headers=request_headers)


		@pytest.mark.parametrize("route", ROUTES, ids=ROUTE_IDS)
		def test_route_rejects_an_anonymous_caller(
		route: tuple[str, str, dict[str, Any] \| None, dict[str, str]],
		) -> None:
		response = _call(_client(), route, token=None)

		assert response.status_code == 401
		assert response.json()["type"] == "urn:oop:ewbi:error:authentication-failed"
		assert response.headers["WWW-Authenticate"] == 'Bearer scope="fed-mgmt"'


		@pytest.mark.parametrize("route", ROUTES, ids=ROUTE_IDS)
		def test_route_rejects_an_unverifiable_token(
		route: tuple[str, str, dict[str, Any] \| None, dict[str, str]],
		) -> None:
		response = _call(_client(), route, token="forged")

		assert response.status_code == 401
		assert response.json()["type"] == "urn:oop:ewbi:error:authentication-failed"


		@pytest.mark.parametrize("route", ROUTES, ids=ROUTE_IDS)
		def test_route_rejects_an_unregistered_partner(
		route: tuple[str, str, dict[str, Any] \| None, dict[str, str]],
		) -> None:
		response = _call(_client(), route, token=STRANGER_TOKEN)

		assert response.status_code == 401
		assert response.json()["type"] == "urn:oop:ewbi:error:partner-unknown"
		assert "partner-nobody" not in response.text


		@pytest.mark.parametrize("route", ROUTES, ids=ROUTE_IDS)
		def test_route_rejects_a_suspended_partner(
		route: tuple[str, str, dict[str, Any] \| None, dict[str, str]],
		) -> None:
		response = _call(_client(partner_status="suspended"), route, token=TOKEN)

		assert response.status_code == 403
		assert response.json()["type"] == "urn:oop:ewbi:error:partner-not-active"

tests/test_federation_health.py

+0 −36

Original line number	Diff line number	Diff line
		@@ -93,45 +93,9 @@ def test_unknown_context_is_404() -> None:
		assert r.json()["type"] == "urn:oop:ewbi:error:federation-context-unknown"


		def test_missing_authorization_header_is_401() -> None:
		r = _client().get(URL)

		assert r.status_code == 401
		assert r.json()["type"] == "urn:oop:ewbi:error:authentication-failed"
		assert r.headers["WWW-Authenticate"] == 'Bearer scope="fed-mgmt"'


		def test_unverifiable_token_is_401() -> None:
		r = _client().get(URL, headers=_bearer("forged"))

		assert r.status_code == 401
		assert r.json()["type"] == "urn:oop:ewbi:error:authentication-failed"


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

		assert r.status_code == 401
		assert r.json()["type"] == "urn:oop:ewbi:error:partner-unknown"


		def test_suspended_partner_is_403() -> None:
		r = _client(partner_status="suspended").get(URL, headers=_bearer(TOKEN))

		assert r.status_code == 403
		assert r.json()["type"] == "urn:oop:ewbi:error:partner-not-active"


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

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


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

		assert "partner-nobody" not in r.text
		assert "partner_ops" not in r.text

tests/test_install_app.py

+0 −9

Original line number	Diff line number	Diff line
		@@ -216,15 +216,6 @@ def test_missing_idempotency_key_is_422() -> None:
		assert publisher.published == []


		def test_missing_token_is_401() -> None:
		client, _, publisher, _ = _client()

		r = client.post(URL, json=_body(), headers={"Idempotency-Key": "idem-1"})

		assert r.status_code == 401
		assert publisher.published == []


		def test_incomplete_body_is_422() -> None:
		client, _, publisher, _ = _client()

tests/test_internal_federation.py

+0 −20

Original line number	Diff line number	Diff line
		@@ -225,26 +225,6 @@ def test_missing_federation_context_is_409() -> None:
		assert ewbi.calls == []


		def test_suspended_partner_is_403() -> None:
		partner = _partner(status="suspended")
		client, _, _ = _client(partner, _agreement(partner))

		r = client.post(URL, json=_body())

		assert r.status_code == 403
		assert r.json()["type"] == "urn:oop:ewbi:error:partner-not-active"


		def test_api_outside_agreement_is_403_violation() -> None:
		partner = _partner()
		client, _, _ = _client(partner, _agreement(partner, api_types={"device-status-retrieve"}))

		r = client.post(URL, json=_body())

		assert r.status_code == 403
		assert r.json()["type"] == "urn:oop:ewbi:error:agreement-violation"


		def test_expired_agreement_is_403_expired() -> None:
		partner = _partner()
		agreement = _agreement(partner)