diff --git a/src/federation_manager/api/errors.py b/src/federation_manager/api/errors.py index 059febac1396fd7d73a5266a91723a91238ce848..e2f0c9ea88e135db987924ad513d2b0272892c05 100644 --- a/src/federation_manager/api/errors.py +++ b/src/federation_manager/api/errors.py @@ -23,6 +23,7 @@ from federation_manager.domain.errors import ( PartnerTokenConfigurationError, PartnerTokenRequestFailed, PartnerUnknown, + SessionUnknown, SrmQueryFailed, UnsupportedServiceApi, problem_type, @@ -240,6 +241,16 @@ def register_exception_handlers(app: FastAPI) -> None: request.url.path, ) + @app.exception_handler(SessionUnknown) + async def _session_unknown(request: Request, exc: SessionUnknown) -> JSONResponse: + return problem( + 404, + "session-unknown", + "Unknown Service API Session", + "No Service API session with this identifier exists for the calling partner.", + request.url.path, + ) + @app.exception_handler(FederationContextUnknown) async def _context_unknown(request: Request, exc: FederationContextUnknown) -> JSONResponse: return problem( diff --git a/src/federation_manager/api/ewbi/v1/service.py b/src/federation_manager/api/ewbi/v1/service.py index 99a265de7b4dd3cb23f015b9c9940fda40871096..b7729c759cd4be98eecae442e81e60a07318f6bd 100644 --- a/src/federation_manager/api/ewbi/v1/service.py +++ b/src/federation_manager/api/ewbi/v1/service.py @@ -1,12 +1,21 @@ from typing import Annotated +from uuid import UUID from fastapi import APIRouter, Depends, Path from federation_manager.api.errors import EWBI_ERROR_RESPONSES from federation_manager.api.security import get_bearer_token +from federation_manager.application.capabilities import QUALITY_ON_DEMAND, InboundCapabilityService from federation_manager.application.queries import InboundQueryService -from federation_manager.contracts.ewbi import ApiForwardingRequest, ServiceApiResponse -from federation_manager.dependencies import get_inbound_query_service +from federation_manager.contracts.ewbi import ( + ApiForwardingRequest, + ServiceApiResponse, + SessionContext, +) +from federation_manager.dependencies import ( + get_inbound_capability_service, + get_inbound_query_service, +) from federation_manager.domain.ewbi import EWBI_BASE_PATH router = APIRouter(prefix=EWBI_BASE_PATH, tags=["ServiceAPIManagement"]) @@ -23,7 +32,40 @@ async def api_forwarding( federationContextId: FederationContextIdPath, # noqa: N803 - GSMA path template name serviceAPINameVal: str, # noqa: N803 - GSMA path template name body: ApiForwardingRequest, - service: Annotated[InboundQueryService, Depends(get_inbound_query_service)], + queries: Annotated[InboundQueryService, Depends(get_inbound_query_service)], + capabilities: Annotated[InboundCapabilityService, Depends(get_inbound_capability_service)], token: Annotated[str, Depends(get_bearer_token)], ) -> ServiceApiResponse: - return await service.forward(token, federationContextId, serviceAPINameVal, body) + if serviceAPINameVal == QUALITY_ON_DEMAND: + return await capabilities.open_session(token, federationContextId, body) + return await queries.forward(token, federationContextId, serviceAPINameVal, body) + + +@router.get( + "/{federationContextId}/apiservice/connid/{connectID}/custid/{customerID}", + operation_id="GetServiceAPISessionInfo", + responses=EWBI_ERROR_RESPONSES, +) +async def get_service_api_session_info( + federationContextId: FederationContextIdPath, # noqa: N803 - GSMA path template name + connectID: str, # noqa: N803 - GSMA path template name + customerID: UUID, # noqa: N803 - GSMA path template name + service: Annotated[InboundCapabilityService, Depends(get_inbound_capability_service)], + token: Annotated[str, Depends(get_bearer_token)], +) -> SessionContext: + return await service.session_info(token, federationContextId, connectID, customerID) + + +@router.delete( + "/{federationContextId}/apiservice/connid/{connectID}/custid/{customerID}", + operation_id="RemoveServiceAPISession", + responses=EWBI_ERROR_RESPONSES, +) +async def remove_service_api_session( + federationContextId: FederationContextIdPath, # noqa: N803 - GSMA path template name + connectID: str, # noqa: N803 - GSMA path template name + customerID: UUID, # noqa: N803 - GSMA path template name + service: Annotated[InboundCapabilityService, Depends(get_inbound_capability_service)], + token: Annotated[str, Depends(get_bearer_token)], +) -> SessionContext: + return await service.close_session(token, federationContextId, connectID, customerID) diff --git a/src/federation_manager/application/capabilities.py b/src/federation_manager/application/capabilities.py new file mode 100644 index 0000000000000000000000000000000000000000..3091fd03238f01f1d1ff4c7b3e6249caf4de46c7 --- /dev/null +++ b/src/federation_manager/application/capabilities.py @@ -0,0 +1,249 @@ +from collections.abc import Callable +from datetime import datetime, timezone +from uuid import UUID, uuid4 + +from federation_manager.application.authentication import PartnerAuthenticator +from federation_manager.application.authorization import AgreementChecker +from federation_manager.contracts.ewbi import ( + ApiForwardingRequest, + ExpiryInterval, + ServiceApiResponse, + SessionContext, + TargetUserContext, +) +from federation_manager.contracts.srm import ( + SUBJECT_CAPABILITY_ACTIVATE, + SUBJECT_CAPABILITY_DEACTIVATE, + CapabilityParameters, + CapabilityTarget, + DeviceTarget, + NetworkCapabilityDeactivatePayloadV1, + NetworkCapabilityPayloadV1, + SourceSpecificationV1, + SrmNetworkCapabilityActivateV1, + SrmNetworkCapabilityDeactivateV1, +) +from federation_manager.domain.errors import ( + AgreementViolation, + FederationContextUnknown, + SessionUnknown, +) +from federation_manager.domain.models import ( + FederationContext, + FederationTransaction, + PartnerOP, +) +from federation_manager.domain.ports import ( + DataBusPublisherPort, + FederationContextRepositoryPort, + TransactionRepositoryPort, +) + +INBOUND = "inbound" +QUALITY_ON_DEMAND = "QualityOnDemand" +API_TYPE = "qod-session" +DEACTIVATE_API_TYPE = "qod-session-remove" +CAPABILITY_TYPE = "qod_session" +DEFAULT_DURATION_SECONDS = 3600 +# expiryInterval caps numHours at 23, so a session longer than this cannot be expressed back +MAX_DURATION_SECONDS = 86399 +OPG_SOURCE = SourceSpecificationV1(family="gsma_opg", api="APIForwarding", version="1.4.0") + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +class InboundCapabilityService: + def __init__( + self, + authenticator: PartnerAuthenticator, + contexts: FederationContextRepositoryPort, + agreements: AgreementChecker, + transactions: TransactionRepositoryPort, + publisher: DataBusPublisherPort, + *, + clock: Callable[[], datetime] = _utcnow, + id_factory: Callable[[], UUID] = uuid4, + ) -> None: + self._authenticator = authenticator + self._contexts = contexts + self._agreements = agreements + self._transactions = transactions + self._publisher = publisher + self._clock = clock + self._new_id = id_factory + + async def open_session( + self, token: str, federation_context_id: str, request: ApiForwardingRequest + ) -> ServiceApiResponse: + partner, context = await self._partner_context(token, federation_context_id) + agreement = await self._agreements.require(partner, API_TYPE) + specification_id = agreement.resolve_api_spec(API_TYPE) + if specification_id is None: + raise AgreementViolation + + content = request.service_api_body.api_content + session_id = self._new_id() + operation_id = self._new_id() + correlation_id = self._new_id() + duration = _duration_seconds(content) + + transaction = FederationTransaction( + id=self._new_id(), + partner_op_id=partner.id, + agreement_id=agreement.id, + federation_context_row_id=context.id, + direction=INBOUND, + operation_id=operation_id, + correlation_id=correlation_id, + api_type=API_TYPE, + status="pending", + external_resource_id=session_id.hex, + external_txn_id=request.txn_identifier, + request_summary={ + "customer_id": str(request.customer_id), + "duration_seconds": duration, + "federation_context_id": federation_context_id, + "service_specification_id": str(specification_id), + }, + started_at=self._clock(), + ) + await self._transactions.add(transaction) + + command = SrmNetworkCapabilityActivateV1( + operation_id=operation_id, + correlation_id=str(correlation_id), + requested_at=self._clock(), + app_provider_id=str(partner.id), + federation_partner_ref=partner.mcc_mnc, + source="federation", + service_instance_id=session_id, + service_specification_id=specification_id, + network_capability=NetworkCapabilityPayloadV1( + capability_type=CAPABILITY_TYPE, + target=CapabilityTarget(device=_device(content)), + parameters=CapabilityParameters(duration_seconds=duration), + profile_ref=_text(content.get("qosProfile")), + source_spec=OPG_SOURCE, + ), + ) + await self._publisher.publish(SUBJECT_CAPABILITY_ACTIVATE, command.model_dump(mode="json")) + await self._transactions.mark_in_progress(transaction.id) + + return ServiceApiResponse( + customer_id=request.customer_id, + txn_identifier=request.txn_identifier, + target_user_context=TargetUserContext( + connect_id=session_id.hex, expiry_duration=_expiry(duration) + ), + ) + + async def session_info( + self, token: str, federation_context_id: str, connect_id: str, customer_id: UUID + ) -> SessionContext: + _, session = await self._session(token, federation_context_id, connect_id, customer_id) + duration = int(str(session.request_summary.get("duration_seconds", 0))) + return SessionContext(connect_id=connect_id, expiry_duration=_expiry(duration)) + + async def close_session( + self, token: str, federation_context_id: str, connect_id: str, customer_id: UUID + ) -> SessionContext: + partner, session = await self._session( + token, federation_context_id, connect_id, customer_id + ) + duration = int(str(session.request_summary.get("duration_seconds", 0))) + closing = SessionContext(connect_id=connect_id, expiry_duration=_expiry(duration)) + + already = await self._transactions.find_by_external_resource_id( + partner.id, DEACTIVATE_API_TYPE, connect_id + ) + if already is not None: + return closing + + operation_id = self._new_id() + correlation_id = self._new_id() + specification_id = session.request_summary.get("service_specification_id") + await self._transactions.add( + FederationTransaction( + id=self._new_id(), + partner_op_id=partner.id, + agreement_id=session.agreement_id, + federation_context_row_id=session.federation_context_row_id, + direction=INBOUND, + operation_id=operation_id, + correlation_id=correlation_id, + api_type=DEACTIVATE_API_TYPE, + status="pending", + external_resource_id=connect_id, + request_summary={"customer_id": str(customer_id)}, + started_at=self._clock(), + ) + ) + command = SrmNetworkCapabilityDeactivateV1( + operation_id=operation_id, + correlation_id=str(correlation_id), + requested_at=self._clock(), + app_provider_id=str(partner.id), + federation_partner_ref=partner.mcc_mnc, + source="federation", + service_specification_id=UUID(str(specification_id)) if specification_id else None, + network_capability=NetworkCapabilityDeactivatePayloadV1( + capability_type=CAPABILITY_TYPE, service_instance_id=UUID(connect_id) + ), + ) + await self._publisher.publish( + SUBJECT_CAPABILITY_DEACTIVATE, command.model_dump(mode="json") + ) + return closing + + async def _partner_context( + self, token: str, federation_context_id: str + ) -> tuple[PartnerOP, FederationContext]: + partner = await self._authenticator.authenticate(token) + context = await self._contexts.find_inbound(partner.id, federation_context_id) + if context is None or context.is_terminated(): + raise FederationContextUnknown(partner.id) + return partner, context + + async def _session( + self, token: str, federation_context_id: str, connect_id: str, customer_id: UUID + ) -> tuple[PartnerOP, FederationTransaction]: + partner, _ = await self._partner_context(token, federation_context_id) + session = await self._transactions.find_by_external_resource_id( + partner.id, API_TYPE, connect_id + ) + if session is None or session.request_summary.get("customer_id") != str(customer_id): + raise SessionUnknown(connect_id) + return partner, session + + +def _device(content: dict[str, object]) -> DeviceTarget: + device = content.get("device") + if not isinstance(device, dict): + raise AgreementViolation + target = DeviceTarget( + phone_number=_text(device.get("phoneNumber")), + ipv6=_text(device.get("ipv6Address")), + network_access_id=_text(device.get("networkAccessIdentifier")), + ) + if not any((target.phone_number, target.ipv6, target.network_access_id)): + raise AgreementViolation + return target + + +def _duration_seconds(content: dict[str, object]) -> int: + duration = content.get("duration") + if not isinstance(duration, int) or duration <= 0: + return DEFAULT_DURATION_SECONDS + return min(duration, MAX_DURATION_SECONDS) + + +def _expiry(seconds: int) -> ExpiryInterval: + return ExpiryInterval( + num_hours=seconds // 3600, num_mins=(seconds % 3600) // 60, num_secs=seconds % 60 + ) + + +def _text(value: object) -> str | None: + return value if isinstance(value, str) and value else None diff --git a/src/federation_manager/contracts/ewbi.py b/src/federation_manager/contracts/ewbi.py index 58703f2a6e8e7bc693f892979be7bc0417c1a1d1..a1d4eb6edf0a8460950dfe99cc378b4807c1ec14 100644 --- a/src/federation_manager/contracts/ewbi.py +++ b/src/federation_manager/contracts/ewbi.py @@ -258,3 +258,12 @@ class InstanceStatusCallback(BaseModel): app_instance_info: AppInstanceInfo = Field( validation_alias="appInstanceInfo", serialization_alias="appInstanceInfo" ) + + +class SessionContext(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + connect_id: str = Field(validation_alias="connectID", serialization_alias="connectID") + expiry_duration: ExpiryInterval = Field( + validation_alias="expiryDuration", serialization_alias="expiryDuration" + ) diff --git a/src/federation_manager/contracts/srm.py b/src/federation_manager/contracts/srm.py index 9dec13bdc7ef5274d2b1e5f64d573fe756255f9e..de79c36b9aa5f56699576b118dacbfe12bbc34af 100644 --- a/src/federation_manager/contracts/srm.py +++ b/src/federation_manager/contracts/srm.py @@ -13,6 +13,8 @@ EVENT_STREAM = "OOP_EVENTS" SUBJECT_OPERATION_COMPLETED = "event.srm.operation.completed" SUBJECT_DEPLOY = "command.srm.service.deploy" SUBJECT_TERMINATE = "command.srm.service.terminate" +SUBJECT_CAPABILITY_ACTIVATE = "command.srm.network.capability.activate" +SUBJECT_CAPABILITY_DEACTIVATE = "command.srm.network.capability.deactivate" class CommandEnvelopeV1(BaseModel): @@ -103,6 +105,7 @@ class CapabilityTarget(BaseModel): class CapabilityParameters(BaseModel): max_age_seconds: int | None = None + duration_seconds: int | None = None class SourceSpecificationV1(BaseModel): @@ -126,3 +129,27 @@ class LocationQueryResponseV1(BaseModel): last_location_time: datetime area: dict[str, Any] srm_resolved_identifier: str | None = None + + +class NetworkCapabilityPayloadV1(BaseModel): + capability_type: str + target: CapabilityTarget + parameters: CapabilityParameters + profile_ref: str | None = None + source_spec: SourceSpecificationV1 | None = None + + +class SrmNetworkCapabilityActivateV1(CommandEnvelopeV1): + service_instance_id: UUID + service_specification_id: UUID + network_capability: NetworkCapabilityPayloadV1 + + +class NetworkCapabilityDeactivatePayloadV1(BaseModel): + capability_type: str + service_instance_id: UUID + + +class SrmNetworkCapabilityDeactivateV1(CommandEnvelopeV1): + network_capability: NetworkCapabilityDeactivatePayloadV1 + service_specification_id: UUID | None = None diff --git a/src/federation_manager/dependencies.py b/src/federation_manager/dependencies.py index b2a6dd8ceb30840b066fbb5a4fe849aba48aa425..0173b9716d51b413ab06efe2d074f0aaebbc6fbb 100644 --- a/src/federation_manager/dependencies.py +++ b/src/federation_manager/dependencies.py @@ -13,6 +13,7 @@ from federation_manager.adapters.database.routing_repo import PostgresRoutingRul from federation_manager.adapters.database.transaction_repo import PostgresTransactionRepo from federation_manager.application.authentication import PartnerAuthenticator from federation_manager.application.authorization import AgreementChecker +from federation_manager.application.capabilities import InboundCapabilityService from federation_manager.application.deployment import InboundDeploymentService from federation_manager.application.federation import InboundFederationService, LocalOperator from federation_manager.application.outbound import OutboundFederationService @@ -165,3 +166,19 @@ def get_inbound_query_service( transaction_repo, srm, ) + + +def get_inbound_capability_service( + authenticator: Annotated[PartnerAuthenticator, Depends(get_partner_authenticator)], + context_repo: Annotated[FederationContextRepositoryPort, Depends(get_federation_context_repo)], + agreement_repo: Annotated[AgreementRepositoryPort, Depends(get_agreement_repo)], + transaction_repo: Annotated[TransactionRepositoryPort, Depends(get_transaction_repo)], + publisher: Annotated[DataBusPublisherPort, Depends(get_command_publisher)], +) -> InboundCapabilityService: + return InboundCapabilityService( + authenticator, + context_repo, + AgreementChecker(agreement_repo), + transaction_repo, + publisher, + ) diff --git a/src/federation_manager/domain/errors.py b/src/federation_manager/domain/errors.py index 4270897cd25b5172d677b0d8d2fd8bd7f45bc020..ac210361518315a0d87cafbd1c968825fe2d9fd8 100644 --- a/src/federation_manager/domain/errors.py +++ b/src/federation_manager/domain/errors.py @@ -97,6 +97,12 @@ class AppInstanceUnknown(FederationError): self.app_instance_id = app_instance_id +class SessionUnknown(FederationError): + def __init__(self, connect_id: str) -> None: + super().__init__(f"no service API session {connect_id} for this partner") + self.connect_id = connect_id + + class IdempotencyKeyReused(FederationError): def __init__(self, idempotency_key: str) -> None: super().__init__(f"idempotency key {idempotency_key} was reused with a different request") diff --git a/tests/test_inbound_query.py b/tests/test_inbound_query.py index 79f5c618abeb5e1e569433385cdbb995f0e110d3..462c47d894cc5aa085fcb55ee0ca548c8d2b2071 100644 --- a/tests/test_inbound_query.py +++ b/tests/test_inbound_query.py @@ -9,6 +9,7 @@ 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, @@ -30,6 +31,7 @@ from tests.fakes import ( InMemoryFederationContextRepo, InMemoryPartnerRepo, InMemoryTransactionRepo, + RecordingCommandPublisher, ) CLIENT_ID = "partner-a" @@ -83,6 +85,7 @@ def _client( app.dependency_overrides[get_agreement_repo] = lambda: InMemoryAgreementRepo([agreement]) app.dependency_overrides[get_transaction_repo] = lambda: transactions app.dependency_overrides[get_srm_client] = lambda: srm + app.dependency_overrides[get_command_publisher] = lambda: RecordingCommandPublisher() return TestClient(app), transactions, srm, partner diff --git a/tests/test_qod_session.py b/tests/test_qod_session.py new file mode 100644 index 0000000000000000000000000000000000000000..20c3b051bab2f5ac7c068363f8dc20ce1b26809a --- /dev/null +++ b/tests/test_qod_session.py @@ -0,0 +1,263 @@ +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from datetime import datetime, timezone +from typing import Any, cast +from uuid import UUID, uuid4 + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from federation_manager.contracts.srm import ( + SUBJECT_CAPABILITY_ACTIVATE, + SUBJECT_CAPABILITY_DEACTIVATE, +) +from federation_manager.dependencies import ( + get_agreement_repo, + get_command_publisher, + get_federation_context_repo, + get_jwt_validator, + get_partner_repo, + get_srm_client, + get_transaction_repo, +) +from federation_manager.domain.models import ( + Agreement, + FederationContext, + PartnerOP, + ValidatedClaims, +) +from federation_manager.main import create_app +from tests.fakes import ( + FakeJwtValidator, + FakeSrmClient, + InMemoryAgreementRepo, + InMemoryFederationContextRepo, + InMemoryPartnerRepo, + InMemoryTransactionRepo, + RecordingCommandPublisher, +) + +CLIENT_ID = "partner-a" +TOKEN = "token-partner-a" +CONTEXT_ID = "fed-ctx-1" +SPEC_ID = uuid4() +CUSTOMER_ID = str(uuid4()) +MSISDN = "+34612345678" +FORWARD_URL = f"/operatorplatform/federation/v1/{CONTEXT_ID}/apiservice/QualityOnDemand" + + +@asynccontextmanager +async def _no_infra(app: FastAPI) -> AsyncIterator[None]: + yield + + +def _client( + permitted: set[str] | None = None, +) -> tuple[TestClient, InMemoryTransactionRepo, RecordingCommandPublisher, PartnerOP]: + partner = PartnerOP(id=uuid4(), mcc_mnc="214-07", oauth2_client_id=CLIENT_ID, status="active") + 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=permitted if permitted is not None else {"qod-session"}, + permitted_zone_ids=set(), + app_mappings=(), + api_family_mappings={"qod-session": SPEC_ID}, + valid_from=datetime(2026, 1, 1, tzinfo=timezone.utc), + valid_until=None, + status="active", + ) + transactions = InMemoryTransactionRepo() + publisher = RecordingCommandPublisher() + + app = create_app(lifespan=_no_infra) + app.dependency_overrides[get_partner_repo] = lambda: InMemoryPartnerRepo([partner]) + app.dependency_overrides[get_jwt_validator] = lambda: FakeJwtValidator( + {TOKEN: ValidatedClaims(client_id=CLIENT_ID, scopes={"fed-mgmt"})} + ) + 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: transactions + app.dependency_overrides[get_command_publisher] = lambda: publisher + app.dependency_overrides[get_srm_client] = lambda: FakeSrmClient() + return TestClient(app), transactions, publisher, partner + + +def _headers() -> dict[str, str]: + return {"Authorization": f"Bearer {TOKEN}"} + + +def _body(**content: Any) -> dict[str, Any]: + api_content: dict[str, Any] = { + "device": {"phoneNumber": MSISDN}, + "qosProfile": "QOS_L", + "duration": 3600, + } + api_content.update(content) + return { + "apiServiceId": "QualityOnDemand", + "customerID": CUSTOMER_ID, + "customerInfo": "ACME Corp", + "txnIdentifier": "partner-txn-qod", + "ServiceAPIBody": {"mediaType": "application/json", "APIContent": api_content}, + } + + +def _session_url(connect_id: str, customer_id: str = CUSTOMER_ID) -> str: + return ( + f"/operatorplatform/federation/v1/{CONTEXT_ID}" + f"/apiservice/connid/{connect_id}/custid/{customer_id}" + ) + + +def _open(client: TestClient) -> str: + response = client.post(FORWARD_URL, json=_body(), headers=_headers()) + assert response.status_code == 200 + return cast(str, response.json()["targetUserContext"]["connectID"]) + + +def test_opening_a_session_publishes_the_activate_command() -> None: + client, transactions, publisher, partner = _client() + + response = client.post(FORWARD_URL, json=_body(), headers=_headers()) + + assert response.status_code == 200 + answered = response.json() + context = answered["targetUserContext"] + assert context["expiryDuration"] == {"numHours": 1, "numMins": 0, "numSecs": 0} + assert answered["txnIdentifier"] == "partner-txn-qod" + assert answered.get("apiResponse") is None + + subject, command = publisher.published[0] + assert subject == SUBJECT_CAPABILITY_ACTIVATE + assert command["source"] == "federation" + assert command["app_provider_id"] == str(partner.id) + assert command["service_specification_id"] == str(SPEC_ID) + assert UUID(cast(str, command["service_instance_id"])).hex == context["connectID"] + capability = cast(dict[str, Any], command["network_capability"]) + assert capability["capability_type"] == "qod_session" + assert capability["profile_ref"] == "QOS_L" + assert capability["parameters"]["duration_seconds"] == 3600 + assert capability["target"]["device"]["phone_number"] == MSISDN + assert capability["source_spec"]["family"] == "gsma_opg" + + audit = transactions.single() + assert audit.api_type == "qod-session" + assert audit.status == "in_progress" + assert audit.external_resource_id == context["connectID"] + + +def test_a_session_without_a_duration_gets_the_default() -> None: + client, _, publisher, _ = _client() + body = _body() + del body["ServiceAPIBody"]["APIContent"]["duration"] + + response = client.post(FORWARD_URL, json=body, headers=_headers()) + + assert response.json()["targetUserContext"]["expiryDuration"]["numHours"] == 1 + capability = cast(dict[str, Any], publisher.published[0][1]["network_capability"]) + assert capability["parameters"]["duration_seconds"] == 3600 + + +def test_a_session_longer_than_a_day_is_clamped() -> None: + client, _, publisher, _ = _client() + + response = client.post(FORWARD_URL, json=_body(duration=172800), headers=_headers()) + + expiry = response.json()["targetUserContext"]["expiryDuration"] + assert (expiry["numHours"], expiry["numMins"], expiry["numSecs"]) == (23, 59, 59) + capability = cast(dict[str, Any], publisher.published[0][1]["network_capability"]) + assert capability["parameters"]["duration_seconds"] == 86399 + + +def test_session_info_is_readable_by_the_owning_partner() -> None: + client, _, _, _ = _client() + connect_id = _open(client) + + response = client.get(_session_url(connect_id), headers=_headers()) + + assert response.status_code == 200 + assert response.json() == { + "connectID": connect_id, + "expiryDuration": {"numHours": 1, "numMins": 0, "numSecs": 0}, + } + + +def test_session_info_for_another_customer_is_404() -> None: + client, _, _, _ = _client() + connect_id = _open(client) + + response = client.get(_session_url(connect_id, str(uuid4())), headers=_headers()) + + assert response.status_code == 404 + assert response.json()["type"] == "urn:oop:ewbi:error:session-unknown" + + +def test_removing_a_session_publishes_the_deactivate_command() -> None: + client, transactions, publisher, _ = _client() + connect_id = _open(client) + + response = client.delete(_session_url(connect_id), headers=_headers()) + + assert response.status_code == 200 + assert response.json()["connectID"] == connect_id + subject, command = publisher.published[1] + assert subject == SUBJECT_CAPABILITY_DEACTIVATE + capability = cast(dict[str, Any], command["network_capability"]) + assert capability["capability_type"] == "qod_session" + assert UUID(cast(str, capability["service_instance_id"])).hex == connect_id + assert command["service_specification_id"] == str(SPEC_ID) + + removals = [t for t in transactions.transactions.values() if t.api_type == "qod-session-remove"] + assert len(removals) == 1 + + +def test_removing_a_session_twice_does_not_republish() -> None: + client, _, publisher, _ = _client() + connect_id = _open(client) + client.delete(_session_url(connect_id), headers=_headers()) + + second = client.delete(_session_url(connect_id), headers=_headers()) + + assert second.status_code == 200 + assert len(publisher.published) == 2 + + +def test_removing_an_unknown_session_is_404() -> None: + client, _, publisher, _ = _client() + _open(client) + + response = client.delete(_session_url(uuid4().hex), headers=_headers()) + + assert response.status_code == 404 + assert len(publisher.published) == 1 + + +def test_qod_outside_the_agreement_is_403() -> None: + client, _, publisher, _ = _client(permitted={"device-location"}) + + response = client.post(FORWARD_URL, json=_body(), headers=_headers()) + + assert response.status_code == 403 + assert response.json()["type"] == "urn:oop:ewbi:error:agreement-violation" + assert publisher.published == [] + + +def test_a_session_request_without_a_device_is_403() -> None: + client, _, publisher, _ = _client() + body = _body() + body["ServiceAPIBody"]["APIContent"]["device"] = {} + + response = client.post(FORWARD_URL, json=body, headers=_headers()) + + assert response.status_code == 403 + assert publisher.published == []