feat: add tests for traffic influence and qod policy deletion and status event handling (34ae9765) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+23 −0

Original line number	Diff line number	Diff line
		@@ -300,6 +300,19 @@ class QualityOnDemandService:
		)
		return

		# A DELETE issued before the activate completed is valid (ADR-0034), so this
		# completion can arrive after teardown already started. The requested delete
		# wins: without this the row would go back to AVAILABLE, the repeat DELETE would
		# republish a deactivate, and the caller would be notified their deleted session
		# is live again.
		if qod_session.state in _TERMINAL_STATES:
		logger.info(
		"activate_completion_ignored_for_terminal_qod_session",
		operation_id=event.operation_id,
		state=qod_session.state.value,
		)
		return

		instance = event.instances[0] if event.instances else None
		update: dict[str, object]
		qos_status: Literal["AVAILABLE", "UNAVAILABLE"]
		@@ -359,6 +372,16 @@ class QualityOnDemandService:
		if qod_session is None:
		return

		# Same rule as the activate completion above: a requested teardown outranks a
		# late backend status signal, which would otherwise revive a deleted session.
		if qod_session.state in _TERMINAL_STATES:
		logger.info(
		"status_change_ignored_for_terminal_qod_session",
		operation_id=event.operation_id,
		state=qod_session.state.value,
		)
		return

		qos_status = (event.metadata or {}).get("qos_status")
		if qos_status not in ("AVAILABLE", "UNAVAILABLE"):
		logger.warning(

src/open_exposure_gateway/application/services/traffic_influence_service.py

+15 −0

Original line number	Diff line number	Diff line
		@@ -398,6 +398,13 @@ class TrafficInfluenceService:
		operation_id=event.operation_id,
		)
		return
		if traffic_influence.state in _TERMINAL_STATES:
		logger.info(
		"activate_completion_ignored_for_terminal_traffic_influence",
		operation_id=event.operation_id,
		state=traffic_influence.state.value,
		)
		return
		update = (
		{"state": RecordTrafficInfluenceState.ACTIVE, "external_ref": instance.external_ref}
		if succeeded and instance is not None
		@@ -470,6 +477,14 @@ class TrafficInfluenceService:
		)
		return

		if traffic_influence.state in _TERMINAL_STATES:
		logger.info(
		"status_change_ignored_for_terminal_traffic_influence",
		operation_id=event.operation_id,
		state=traffic_influence.state.value,
		)
		return

		state = (
		RecordTrafficInfluenceState.ACTIVE
		if qos_status == "AVAILABLE"

tests/unit/test_qod_service.py

+61 −0

Original line number	Diff line number	Diff line
		@@ -371,6 +371,40 @@ class TestHandleCompleted:
		)
		return operation, qod_session

		async def test_activate_completion_after_delete_does_not_revive_the_session(
		self,
		service: QualityOnDemandService,
		operation_repo: FakeOperationRepository,
		qod_session_repo: FakeQodSessionRepository,
		) -> None:
		"""ADR-0034: a DELETE issued before the activate completes is valid, so the
		activate completion legitimately arrives mid-teardown. The delete must win --
		otherwise the session returns to AVAILABLE and a repeat DELETE republishes a
		deactivate instead of 404ing."""
		operation, qod_session = await self._seed(operation_repo, qod_session_repo)
		await qod_session_repo.save(
		qod_session.model_copy(update={"state": QodSessionState.DELETION_REQUESTED})
		)
		event = SRMOperationCompleted.model_validate(
		completion_payload(
		str(operation.operation_id),
		instances=[
		{
		"service_instance_id": str(uuid4()),
		"zone_id": str(uuid4()),
		"status": "completed",
		"external_ref": "qod-session-123",
		}
		],
		)
		)

		await service.handle_completed(event)

		updated = await qod_session_repo.get_by_id(qod_session.session_id)
		assert updated is not None
		assert updated.state == QodSessionState.DELETION_REQUESTED

		async def test_completed_event_marks_session_available(
		self,
		service: QualityOnDemandService,
		@@ -856,6 +890,33 @@ class TestHandleStatusChanged:
		)
		return operation, qod_session

		async def test_status_event_after_delete_does_not_revive_the_session(
		self,
		service: QualityOnDemandService,
		operation_repo: FakeOperationRepository,
		qod_session_repo: FakeQodSessionRepository,
		callback_delivery_port: FakeQodCallbackDeliveryPort,
		) -> None:
		"""The activate operation stays live until its deactivate completes, so a backend
		status signal can still arrive mid-teardown. Same rule as the completion event:
		the requested delete wins, and no qos-status-changed is sent."""
		operation, qod_session = await self._seed_available(operation_repo, qod_session_repo)
		await qod_session_repo.save(
		qod_session.model_copy(update={"state": QodSessionState.DELETION_REQUESTED})
		)
		event = SRMOperationStatus.model_validate(
		operation_status_payload(
		str(operation.operation_id), metadata={"qos_status": "AVAILABLE"}
		)
		)

		await service.handle_status_changed(event)

		updated = await qod_session_repo.get_by_id(qod_session.session_id)
		assert updated is not None
		assert updated.state == QodSessionState.DELETION_REQUESTED
		assert callback_delivery_port.delivered == []

		@pytest.mark.parametrize(
		("metadata", "expected_status_info"),
		[

tests/unit/test_traffic_influence_flows.py

+79 −0

Original line number	Diff line number	Diff line
		@@ -334,6 +334,85 @@ class TestTrafficInfluenceCreateFlow:
		traffic_influence_repo.rows[traffic_influence_id].state == TrafficInfluenceState.ACTIVE
		)

		async def test_activate_completion_after_delete_does_not_revive_the_policy(
		self,
		api_client: TestClient,
		live_ti: None,
		fake_bus: FakeDataBus,
		traffic_influence_repo: FakeTrafficInfluenceRepository,
		traffic_influence_callback_delivery_port: FakeTrafficInfluenceCallbackDeliveryPort,
		) -> None:
		"""ADR-0034: a DELETE issued before the activate completes is valid, so the
		activate completion legitimately arrives after teardown started. The delete must
		win -- otherwise the row returns to 'active', the repeat DELETE republishes a
		deactivate instead of 404ing, and the subscriber is told the policy they deleted
		is live."""
		body = {**TI_BODY, "subscriptionRequest": SUBSCRIPTION_REQUEST}
		response = api_client.post(f"{TI_BASE}/traffic-influences", json=body)
		traffic_influence_id = UUID(response.json()["trafficInfluenceID"])
		activate_operation_id = traffic_influence_repo.rows[traffic_influence_id].operation_id

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

		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",
		}
		],
		),
		)

		assert (
		traffic_influence_repo.rows[traffic_influence_id].state
		== TrafficInfluenceState.DELETION_IN_PROGRESS
		)
		assert (
		api_client.get(f"{TI_BASE}/traffic-influences/{traffic_influence_id}").json()["state"]
		== "deletion in progress"
		)
		assert (
		api_client.delete(f"{TI_BASE}/traffic-influences/{traffic_influence_id}").status_code
		== 404
		)
		assert len([p for s, p in fake_bus.published if s == Subject.TASK_DEACTIVATE]) == 1
		assert traffic_influence_callback_delivery_port.delivered == []

		async def test_status_event_after_delete_does_not_revive_the_policy(
		self,
		api_client: TestClient,
		live_ti: None,
		fake_bus: FakeDataBus,
		traffic_influence_repo: FakeTrafficInfluenceRepository,
		) -> None:
		"""The activate operation stays live until its deactivate completes, so a backend
		status signal can still arrive mid-teardown. Same rule as the completion event:
		the requested delete wins."""
		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

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

		await fake_bus.publish(
		str(Subject.OPERATION_STATUS),
		operation_status_payload(str(operation_id), metadata={"qos_status": "AVAILABLE"}),
		)

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

		async def test_redelivered_completion_event_is_ignored(
		self,
		api_client: TestClient,