feat: implement stubs for traffic influence device creation and updates,... (633602f8) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/api/camara/traffic_influence/vwip/router.py

+30 −0

Original line number	Diff line number	Diff line
		@@ -5,7 +5,9 @@ from fastapi import APIRouter, Depends, Query, status

		from open_exposure_gateway.api.camara.common import IdempotencyKeyHeader, XCorrelatorHeader
		from open_exposure_gateway.api.camara.traffic_influence.vwip.schemas import (
		PatchTrafficInfluence,
		PostTrafficInfluence,
		PostTrafficInfluenceDevice,
		TrafficInfluence,
		)
		from open_exposure_gateway.application.services.traffic_influence_service import (
		@@ -17,6 +19,7 @@ from open_exposure_gateway.core.exceptions import (
		DownstreamServiceException,
		ForbiddenException,
		NotFoundException,
		NotImplementedException,
		UnauthorizedException,
		)
		from open_exposure_gateway.dependencies import (
		@@ -47,6 +50,7 @@ _ERROR_RESPONSES: dict[int \| str, dict[str, Any]] = {
		ForbiddenException,
		NotFoundException,
		ConflictException,
		NotImplementedException,
		DownstreamServiceException,
		)
		}
		@@ -140,3 +144,29 @@ async def delete_traffic_influence(
		app_provider_id=caller.app_provider_id,
		x_correlator=x_correlator,
		)


		@router.post(
		"/traffic-influence-devices",
		tags=["Traffic Influence API write"],
		summary="Creates a new TrafficInfluence resource influencing the traffic "
		"toward local instances of the Application for a specific user.",
		responses=_responses(400, 401, 403, 404, 409, 500, 501, 503),
		)
		async def post_traffic_influence_device(request: PostTrafficInfluenceDevice) -> Any:
		raise NotImplementedException(
		message="Device-scoped traffic influence is not implemented in this release"
		)


		@router.patch(
		"/traffic-influences/{trafficInfluenceID}",
		tags=["Traffic Influence API write"],
		summary="Updates a specific TrafficInfluence resource, identified by the "
		"trafficInfluenceID value.",
		responses=_responses(400, 401, 403, 404, 409, 500, 501, 503),
		)
		async def patch_traffic_influence(trafficInfluenceID: str, request: PatchTrafficInfluence) -> Any:
		raise NotImplementedException(
		message="Traffic influence modification is not implemented in this release"
		)

src/open_exposure_gateway/api/camara/traffic_influence/vwip/schemas.py

+31 −0

Original line number	Diff line number	Diff line
		@@ -72,6 +72,37 @@ class PostTrafficInfluence(BaseTrafficInfluence):
		pass


		class Device(BaseModel):
		"""CAMARA `Device` -- the subscriber a device-scoped policy applies to. Modelled
		loosely: `POST /traffic-influence-devices` is a 501 stub, so nothing reads these
		fields yet. Tighten to the full DeviceIpv4Addr/PhoneNumber shapes when it is served.
		"""

		phoneNumber: Optional[str] = None
		networkAccessIdentifier: Optional[str] = None
		ipv4Address: Optional[dict[str, Any]] = None
		ipv6Address: Optional[str] = None


		class PostTrafficInfluenceDevice(BaseTrafficInfluence):
		device: Optional[Device] = None


		class PatchTrafficInfluence(BaseModel):
		"""CAMARA marks `trafficInfluenceID`/`apiConsumerId`/`appId`/`state` readOnly on
		PATCH, so they are absent here. Note the spec still inherits
		`required: [apiConsumerId, appId]` from the base schema while marking both readOnly --
		a contradiction left permissive here rather than resolved, since PATCH is a 501 stub.
		"""

		appInstanceId: Optional[UUID] = None
		edgeCloudRegion: Optional[str] = None
		edgeCloudZoneId: Optional[UUID] = None
		sourceTrafficFilters: Optional[SourceTrafficFilters] = None
		destinationTrafficFilters: Optional[DestinationTrafficFilters] = None
		subscriptionRequest: Optional[SubscriptionRequest] = None


		class TrafficInfluence(BaseTrafficInfluence):
		trafficInfluenceID: UUID
		state: TrafficInfluenceState

tests/conformance/conftest.py

+48 −0

Original line number	Diff line number	Diff line
		@@ -20,16 +20,25 @@ from open_exposure_gateway.application.services.location_retrieval_service impor
		from open_exposure_gateway.application.services.quality_on_demand_service import (
		QualityOnDemandService,
		)
		from open_exposure_gateway.application.services.traffic_influence_service import (
		TrafficInfluenceService,
		)
		from open_exposure_gateway.dependencies import (
		get_edge_app_service,
		get_location_retrieval_service,
		get_publisher,
		get_qod_service,
		get_traffic_influence_service,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		SRMZone,
		SRMZoneMetadata,
		)
		from open_exposure_gateway.domain.models import (
		AppRegistration,
		AppRegistrationStatus,
		PackageType,
		)
		from tests.conformance.harness import app
		from tests.unit.fakes import (
		FakeAppInstanceRepository,
		@@ -41,9 +50,12 @@ from tests.unit.fakes import (
		FakeQodCallbackDeliveryPort,
		FakeQodSessionRepository,
		FakeSRMClient,
		FakeTrafficInfluenceCallbackDeliveryPort,
		FakeTrafficInfluenceRepository,
		wire_operation_consumer,
		wire_qod_operation_consumer,
		wire_srm_worker,
		wire_traffic_influence_operation_consumer,
		)


		@@ -97,6 +109,42 @@ def service_overrides() -> Generator[None, None, None]:
		callback_delivery_repo=qod_callback_delivery_repo,
		callback_delivery_port=qod_callback_delivery_port,
		)
		# Traffic Influence resolves service_specification_id from an already-registered app
		# (ADR-0011), so the catalog cannot be empty the way it can for QoD.
		ti_app_registration_repo = FakeAppRegistrationRepository()
		ti_app_registration_id = uuid4()
		ti_app_registration_repo.rows[ti_app_registration_id] = AppRegistration(
		app_registration_id=ti_app_registration_id,
		app_id=uuid4(),
		tenant_id="conformance-tenant",
		name="conformance-app",
		version="1.0.0",
		package_type=PackageType.HELM,
		status=AppRegistrationStatus.REGISTERED,
		)
		ti_operation_repo = FakeOperationRepository()
		ti_repo = FakeTrafficInfluenceRepository()
		ti_callback_registration_repo = FakeCallbackRegistrationRepository()
		ti_callback_delivery_repo = FakeCallbackDeliveryRepository()
		ti_callback_delivery_port = FakeTrafficInfluenceCallbackDeliveryPort()
		wire_traffic_influence_operation_consumer(
		bus,
		ti_operation_repo,
		ti_repo,
		ti_callback_registration_repo,
		ti_callback_delivery_repo,
		ti_callback_delivery_port,
		)
		app.dependency_overrides[get_traffic_influence_service] = lambda: TrafficInfluenceService(
		srm_client=srm,
		publisher=bus,
		operation_repo=ti_operation_repo,
		traffic_influence_repo=ti_repo,
		callback_registration_repo=ti_callback_registration_repo,
		callback_delivery_repo=ti_callback_delivery_repo,
		callback_delivery_port=ti_callback_delivery_port,
		app_registration_repo=ti_app_registration_repo,
		)
		app.dependency_overrides[get_location_retrieval_service] = lambda: LocationRetrievalService(
		srm_client=srm
		)

tests/conformance/test_traffic_influence_conformance.py

0 → 100644

+84 −0

Original line number	Diff line number	Diff line
		"""CAMARA Traffic Influence conformance.

		Every served operation in the vendored upstream spec is exercised with
		generated requests against the ASGI app; responses are validated against the
		spec (status codes, response schemas, headers). A failure here means the
		northbound interface diverges from CAMARA.
		"""

		from pathlib import Path
		from typing import TYPE_CHECKING
		from uuid import uuid4

		import pytest
		import schemathesis
		from fastapi.testclient import TestClient

		from open_exposure_gateway.api.camara.traffic_influence.vwip.router import BASE_PATH
		from tests.conformance.harness import app

		if TYPE_CHECKING:
		from schemathesis.specs.openapi.schemas import OpenApiCase

		pytestmark = pytest.mark.conformance

		SPEC = (
		Path(__file__).parents[2]
		/ "src"
		/ "open_exposure_gateway"
		/ "api"
		/ "camara"
		/ "traffic_influence"
		/ "vwip"
		/ "API_definitions"
		/ "traffic_influence.yaml"
		)

		# Served as 501 stubs (see the router's KNOWN DEVIATION note). traffic_influence.yaml
		# documents 501 on no operation, so these would fail UndefinedStatusCode; they are
		# excluded so the rest of the surface stays under conformance rather than the whole
		# suite going red. test_stubbed_operations_still_return_501 pins the deviation, so it
		# cannot change without this exclusion being revisited.
		STUBBED = {
		("POST", "/traffic-influence-devices"),
		("PATCH", "/traffic-influences/{trafficInfluenceID}"),
		}

		schema = schemathesis.openapi.from_path(SPEC)
		schema = schema.exclude(method="PATCH", path="/traffic-influences/{trafficInfluenceID}").exclude(
		method="POST", path="/traffic-influence-devices"
		)
		schema.app = app
		schema.config.generation.update(max_examples=10, no_shrink=True)
		# Business rules the JSON Schema cannot express reject otherwise-valid input with 400,
		# which is a documented response on every operation here: `appId` must resolve to a
		# registered app (ADR-0011), and `trafficInfluenceID` is a UUID OEG minted even though
		# CAMARA types the path parameter as a bare string. Same allowance as the QoD suite.
		schema.config.checks.positive_data_acceptance.expected_statuses = [
		*schema.config.checks.positive_data_acceptance.expected_statuses,
		"400",
		]


		@schema.parametrize()
		def test_traffic_influence_conformance(case: "OpenApiCase") -> None:
		case.call_and_validate()


		def test_stubbed_operations_still_return_501() -> None:
		"""Guards the STUBBED exclusion above. These two are excluded from conformance only
		because they answer 501, which this spec documents nowhere. Implement either one --
		or change what it returns -- and this fails, forcing the exclusion to be revisited
		rather than quietly covering a now-conformant operation."""
		client = TestClient(app)
		statuses = {
		("POST", "/traffic-influence-devices"): client.post(
		f"{BASE_PATH}/traffic-influence-devices",
		json={"apiConsumerId": "c1", "appId": str(uuid4())},
		).status_code,
		("PATCH", "/traffic-influences/{trafficInfluenceID}"): client.patch(
		f"{BASE_PATH}/traffic-influences/{uuid4()}", json={}
		).status_code,
		}
		assert set(statuses) == STUBBED
		assert statuses == dict.fromkeys(STUBBED, 501)

tests/unit/test_traffic_influence_camara_schema.py

+1 −0

Original line number	Diff line number	Diff line
		@@ -22,6 +22,7 @@ from tests.unit.fakes import (
		FakeTrafficInfluenceRepository,
		completion_payload,
		)

		# `registered_app` is imported for its autouse effect in this module: every Traffic
		# Influence request targets an already-registered app (ADR-0011).
		from tests.unit.test_traffic_influence_flows import TI_BODY, registered_app # noqa: F401