test(traffic-influence): add flow tests for POST/GET/DELETE (396c0771) · Commits · OOP / code / open-exposure-gateway

tests/unit/test_traffic_influence_flows.py

0 → 100644

+300 −0

Original line number	Diff line number	Diff line
		"""End-to-end flow tests for Traffic Influence.

		Same setup as test_qod_flows.py: real HTTP API against the in-memory fakes.
		A failing test here means a real hole in the flow, not a broken test.
		"""

		from typing import Any
		from uuid import UUID, uuid4

		import pytest
		from fastapi.testclient import TestClient

		from open_exposure_gateway.api.camara.traffic_influence.vwip.router import (
		BASE_PATH as TI_BASE,
		)
		from open_exposure_gateway.domain.models import OperationStatus, TrafficInfluenceState
		from open_exposure_gateway.domain.traffic_influence import (
		SRMNetworkCapabilityActivateCommand,
		SRMNetworkCapabilityDeactivateCommand,
		Subject,
		)
		from tests.unit.fakes import (
		FakeDataBus,
		FakeOperationRepository,
		FakeTrafficInfluenceRepository,
		completion_payload,
		)

		TI_BODY: dict[str, Any] = {
		"apiConsumerId": "consumer-1",
		"appId": "6b29fc40-ca47-1067-b31d-00dd010662da",
		"edgeCloudZoneId": "642f6105-7015-4af1-a4d1-e1ecb8437abc",
		"sourceTrafficFilters": {"sourcePort": 45678},
		"destinationTrafficFilters": {"destinationPort": 443, "destinationProtocol": "TCP"},
		}

		SUBSCRIPTION_REQUEST: dict[str, Any] = {
		"protocol": "HTTP",
		"sink": "https://endpoint.example.com/sink",
		"types": ["org.camaraproject.traffic-influence.v1.traffic-influence-change"],
		"config": {"subscriptionDetail": {}},
		}


		class TestTrafficInfluenceCreateFlow:
		def test_create_returns_201(self, api_client: TestClient) -> None:
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=TI_BODY)
		assert response.status_code == 201
		body = response.json()
		assert body["state"] == "ordered"
		assert body["appId"] == TI_BODY["appId"]
		assert "trafficInfluenceID" in body

		def test_create_accepts_subscription_request(self, api_client: TestClient) -> None:
		body = {**TI_BODY, "subscriptionRequest": SUBSCRIPTION_REQUEST}
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=body)
		assert response.status_code == 201

		def test_srm_receives_a_valid_activate_command(
		self, api_client: TestClient, fake_bus: FakeDataBus
		) -> None:
		api_client.post(f"{TI_BASE}/traffic-influences", json=TI_BODY)

		activates = [p for s, p in fake_bus.published if s == Subject.TASK_ACTIVATE]
		assert len(activates) == 1
		command = SRMNetworkCapabilityActivateCommand.model_validate(activates[0])
		source = command.network_capability.parameters.traffic_filters.source
		destination = command.network_capability.parameters.traffic_filters.destination
		assert source is not None
		assert destination is not None
		assert (
		command.network_capability.target.application_reference.external_app_id
		== TI_BODY["appId"]
		)
		assert source.port == 45678
		assert destination.port == 443
		assert destination.protocol == "TCP"
		assert command.resource_zone_id == TI_BODY["edgeCloudZoneId"]
		assert command.service_specification_id == TI_BODY["appId"]
		assert command.source == "nbi_camara"

		def test_persists_operation_and_traffic_influence_rows(
		self,
		api_client: TestClient,
		operation_repo: FakeOperationRepository,
		traffic_influence_repo: FakeTrafficInfluenceRepository,
		) -> None:
		"""Proves the DI wiring end-to-end: the router/service must reach the
		real repositories, not just the constructor accepting them."""
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=TI_BODY)
		traffic_influence_id = UUID(response.json()["trafficInfluenceID"])

		operations = list(operation_repo.rows.values())
		assert len(operations) == 1
		assert operations[0].status == OperationStatus.PENDING

		stored = traffic_influence_repo.rows.get(traffic_influence_id)
		assert stored is not None
		assert stored.state == TrafficInfluenceState.ORDERED
		assert stored.operation_id == operations[0].operation_id
		assert str(stored.app_id) == TI_BODY["appId"]

		async def test_completion_event_marks_traffic_influence_active(
		self,
		api_client: TestClient,
		live_ti: None,
		fake_bus: FakeDataBus,
		traffic_influence_repo: FakeTrafficInfluenceRepository,
		) -> None:
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=TI_BODY)
		traffic_influence_id = UUID(response.json()["trafficInfluenceID"])
		operation_id = traffic_influence_repo.rows[traffic_influence_id].operation_id

		await fake_bus.publish(
		str(Subject.OPERATION_COMPLETED),
		completion_payload(
		str(operation_id),
		instances=[
		{
		"service_instance_id": str(uuid4()),
		"zone_id": str(uuid4()),
		"status": "completed",
		"external_ref": "traffic-policy-456",
		}
		],
		),
		)

		updated = traffic_influence_repo.rows[traffic_influence_id]
		assert updated.state == TrafficInfluenceState.ACTIVE
		assert updated.external_ref == "traffic-policy-456"


		class TestTrafficInfluenceGetFlow:
		@pytest.mark.skip(
		reason="TODO: GET /traffic-influences/{id} is a synchronous SRM call against an "
		"undocumented endpoint (architecture/srm/interface-contract.md §E has no "
		"network-capability read endpoint), and OEG's minted trafficInfluenceID has no "
		"reconciled mapping to SRM's external_ref -- same open gap as QoD's GET "
		"(test_qod_flows.py). A resource created via POST is therefore not retrievable "
		"via GET until SRM confirms the real wire shape."
		)
		def test_get_traffic_influence_returns_created_resource(self, api_client: TestClient) -> None:
		traffic_influence_id = api_client.post(
		f"{TI_BASE}/traffic-influences", json=TI_BODY
		).json()["trafficInfluenceID"]
		response = api_client.get(f"{TI_BASE}/traffic-influences/{traffic_influence_id}")
		assert response.status_code == 200
		assert response.json()["trafficInfluenceID"] == traffic_influence_id

		def test_get_traffic_influence_returns_404_when_unknown(self, api_client: TestClient) -> None:
		response = api_client.get(f"{TI_BASE}/traffic-influences/{uuid4()}")
		assert response.status_code == 404

		def test_list_traffic_influences_returns_empty_by_default(self, api_client: TestClient) -> None:
		response = api_client.get(f"{TI_BASE}/traffic-influences")
		assert response.status_code == 200
		assert response.json() == []


		class TestTrafficInfluenceDeleteFlow:
		def test_delete_returns_404_before_activation_confirmed(self, api_client: TestClient) -> None:
		"""A resource with no confirmed external_ref yet (still ORDERED) has nothing
		for a deactivate command to key on, so deletion 404s rather than silently
		no-op-ing or guessing at an id SRM never confirmed."""
		traffic_influence_id = api_client.post(
		f"{TI_BASE}/traffic-influences", json=TI_BODY
		).json()["trafficInfluenceID"]

		response = api_client.delete(f"{TI_BASE}/traffic-influences/{traffic_influence_id}")

		assert response.status_code == 404

		def test_delete_returns_404_when_unknown(self, api_client: TestClient) -> None:
		response = api_client.delete(f"{TI_BASE}/traffic-influences/{uuid4()}")
		assert response.status_code == 404

		async def test_delete_publishes_deactivate_command_once_available(
		self,
		api_client: TestClient,
		live_ti: None,
		fake_bus: FakeDataBus,
		traffic_influence_repo: FakeTrafficInfluenceRepository,
		) -> None:
		"""DELETE must actually tear the policy down at SRM: a 202 with no deactivate
		command published would silently leak a traffic-influence policy."""
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=TI_BODY)
		traffic_influence_id = UUID(response.json()["trafficInfluenceID"])
		operation_id = traffic_influence_repo.rows[traffic_influence_id].operation_id

		await fake_bus.publish(
		str(Subject.OPERATION_COMPLETED),
		completion_payload(
		str(operation_id),
		instances=[
		{
		"service_instance_id": str(uuid4()),
		"zone_id": str(uuid4()),
		"status": "completed",
		"external_ref": "traffic-policy-456",
		}
		],
		),
		)

		delete_response = api_client.delete(f"{TI_BASE}/traffic-influences/{traffic_influence_id}")

		assert delete_response.status_code == 202
		deactivates = [p for s, p in fake_bus.published if s == Subject.TASK_DEACTIVATE]
		assert len(deactivates) == 1
		command = SRMNetworkCapabilityDeactivateCommand.model_validate(deactivates[0])
		assert command.network_capability.external_ref == "traffic-policy-456"
		assert (
		traffic_influence_repo.rows[traffic_influence_id].state
		== TrafficInfluenceState.DELETION_IN_PROGRESS
		)

		async def test_delete_completion_event_marks_resource_deleted(
		self,
		api_client: TestClient,
		live_ti: None,
		fake_bus: FakeDataBus,
		traffic_influence_repo: FakeTrafficInfluenceRepository,
		) -> None:
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=TI_BODY)
		traffic_influence_id = UUID(response.json()["trafficInfluenceID"])
		activate_operation_id = traffic_influence_repo.rows[traffic_influence_id].operation_id

		await fake_bus.publish(
		str(Subject.OPERATION_COMPLETED),
		completion_payload(
		str(activate_operation_id),
		instances=[
		{
		"service_instance_id": str(uuid4()),
		"zone_id": str(uuid4()),
		"status": "completed",
		"external_ref": "traffic-policy-456",
		}
		],
		),
		)

		api_client.delete(f"{TI_BASE}/traffic-influences/{traffic_influence_id}")
		deactivates = [p for s, p in fake_bus.published if s == Subject.TASK_DEACTIVATE]
		deactivate_operation_id = deactivates[0]["operation_id"]

		await fake_bus.publish(
		str(Subject.OPERATION_COMPLETED),
		completion_payload(
		deactivate_operation_id,
		instances=[
		{
		"service_instance_id": str(uuid4()),
		"zone_id": str(uuid4()),
		"status": "completed",
		"external_ref": "traffic-policy-456",
		}
		],
		),
		)

		assert (
		traffic_influence_repo.rows[traffic_influence_id].state == TrafficInfluenceState.DELETED
		)


		class TestTrafficInfluenceValidation:
		"""CAMARA traffic_influence.yaml constraints the request schema must enforce (400s)."""

		def test_missing_api_consumer_id_is_rejected(self, api_client: TestClient) -> None:
		body = {k: v for k, v in TI_BODY.items() if k != "apiConsumerId"}
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=body)
		assert response.status_code == 400

		def test_missing_app_id_is_rejected(self, api_client: TestClient) -> None:
		body = {k: v for k, v in TI_BODY.items() if k != "appId"}
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=body)
		assert response.status_code == 400

		def test_source_port_out_of_range_is_rejected(self, api_client: TestClient) -> None:
		body = {**TI_BODY, "sourceTrafficFilters": {"sourcePort": 70000}}
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=body)
		assert response.status_code == 400

		def test_subscription_request_without_sink_is_rejected(self, api_client: TestClient) -> None:
		bad_subscription = {k: v for k, v in SUBSCRIPTION_REQUEST.items() if k != "sink"}
		body = {**TI_BODY, "subscriptionRequest": bad_subscription}
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=body)
		assert response.status_code == 400

		def test_subscription_request_with_non_https_sink_is_rejected(
		self, api_client: TestClient
		) -> None:
		body = {
		**TI_BODY,
		"subscriptionRequest": {**SUBSCRIPTION_REQUEST, "sink": "http://insecure.example.com"},
		}
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=body)
		assert response.status_code == 400

Admin message