test: add unit and flow tests for QOS_STATUS_CHANGED delivery (8337c0d6) · Commits · OOP / code / open-exposure-gateway

tests/unit/fakes.py

+13 −0

Original line number	Diff line number	Diff line
		@@ -178,6 +178,19 @@ def completion_payload(operation_id: str, **overrides: Any) -> dict[str, Any]:
		return payload


		def operation_status_payload(operation_id: str, **overrides: Any) -> dict[str, Any]:
		payload: dict[str, Any] = {
		"schema_version": "1.0",
		"operation_id": operation_id,
		"state": "completed",
		"metadata": {"qos_status": "UNAVAILABLE"},
		"correlation_id": str(uuid4()),
		"emitted_at": "2026-07-03T12:00:00+00:00",
		}
		payload.update(overrides)
		return payload


		def wire_srm_worker(bus: FakeDataBus, srm: FakeSRMClient) -> None:
		async def on_deploy(command: dict[str, Any]) -> None:
		# POST /appinstances always carries exactly one targets[] entry

tests/unit/test_qod_flows.py

+50 −0

Original line number	Diff line number	Diff line
		@@ -20,10 +20,13 @@ from open_exposure_gateway.domain.quality_on_demand import (
		Subject,
		)
		from tests.unit.fakes import (
		FakeCallbackDeliveryRepository,
		FakeDataBus,
		FakeOperationRepository,
		FakeQodCallbackDeliveryPort,
		FakeQodSessionRepository,
		completion_payload,
		operation_status_payload,
		)

		SESSION_BODY: dict[str, Any] = {
		@@ -113,6 +116,53 @@ class TestQodSessionFlow:
		assert updated.state == QodSessionState.AVAILABLE
		assert updated.external_ref == "qod-session-nef-123"

		async def test_webhook_receives_qos_status_changed_on_completion_and_status_events(
		self,
		api_client: TestClient,
		live_qod: None,
		fake_bus: FakeDataBus,
		qod_session_repo: FakeQodSessionRepository,
		callback_delivery_repo: FakeCallbackDeliveryRepository,
		callback_delivery_port: FakeQodCallbackDeliveryPort,
		) -> None:
		body = {**SESSION_BODY, "webhook": {"notificationUrl": "https://client.example.com/cb"}}
		response = api_client.post(f"{QOD_BASE}/sessions", json=body)
		session_id = UUID(response.json()["sessionId"])
		operation_id = qod_session_repo.rows[session_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": "qod-session-nef-123",
		}
		],
		),
		)

		assert len(callback_delivery_port.delivered) == 1
		sink, first_event = callback_delivery_port.delivered[0]
		assert sink == "https://client.example.com/cb"
		assert first_event.data.qosStatus == "AVAILABLE"
		assert qod_session_repo.rows[session_id].state == QodSessionState.AVAILABLE

		await fake_bus.publish(
		str(Subject.OPERATION_STATUS),
		operation_status_payload(str(operation_id)),
		)

		assert len(callback_delivery_port.delivered) == 2
		_, second_event = callback_delivery_port.delivered[1]
		assert second_event.data.qosStatus == "UNAVAILABLE"
		assert second_event.data.statusInfo == "NETWORK_TERMINATED"
		assert qod_session_repo.rows[session_id].state == QodSessionState.UNAVAILABLE
		assert len(callback_delivery_repo.rows) == 2

		@pytest.mark.skip(
		reason="TODO: GET /sessions/{id} is a synchronous SRM call (unchanged by the async "
		"POST /sessions work) and OEG's minted sessionId has no reconciled mapping to SRM's "

tests/unit/test_qod_service.py

+496 −3

File changed.

Preview size limit exceeded, changes collapsed.

Admin message