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

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

+30 −0

Changes for src/open_exposure_gateway/api/camara/traffic_influence/vwip/router.py: 30 added lines, 0 removed lines.
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

Changes for src/open_exposure_gateway/api/camara/traffic_influence/vwip/schemas.py: 31 added lines, 0 removed lines.
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

Changes for tests/conformance/conftest.py: 48 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -20,17 +20,26 @@ 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_caller_context,
		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.conftest import fake_caller_context
		from tests.unit.fakes import (
		@@ -44,9 +53,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,
		)


		@@ -104,6 +116,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

Changes for tests/conformance/test_traffic_influence_conformance.py: 84 added lines, 0 removed lines.
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

Changes for tests/unit/test_traffic_influence_camara_schema.py: 1 added line, 0 removed lines.
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