fix(TI): forward pinned appInstanceId and report SRM's selected instance (9a02ef1f) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/mappers/traffic_influence_mapper.py

+1 −0

Changes for src/open_exposure_gateway/application/mappers/traffic_influence_mapper.py: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -62,6 +62,7 @@ def build_activate_command(
		application_reference=NetworkCapabilityTargetApplicationReference(
		service_specification_ref=str(request.appId),
		external_app_id=str(request.appId),
		app_instance_id=str(request.appInstanceId) if request.appInstanceId else None,
		),
		device=NetworkCapabilityTargetDevice(
		phone_number=device.phoneNumber,

src/open_exposure_gateway/application/services/traffic_influence_service.py

+42 −13

Changes for src/open_exposure_gateway/application/services/traffic_influence_service.py: 42 added lines, 13 removed lines.
Original line number	Diff line number	Diff line
		@@ -97,6 +97,25 @@ _TEARDOWN_STATES = frozenset(
		)


		def _reported_selected_app_instance_ids(operation: Optional[Operation]) -> list[UUID] \| None:
		if operation is None or not operation.result:
		return None
		for instance in operation.result.get("instances", []):
		reported = (instance.get("metadata") or {}).get("selected_app_instance_ids")
		if reported is None:
		continue
		try:
		return [UUID(i) for i in reported]
		except (TypeError, ValueError):
		logger.warning(
		"srm_selected_app_instance_ids_malformed",
		operation_id=str(operation.operation_id),
		selected_app_instance_ids=reported,
		)
		return None
		return None


		class TrafficInfluenceService:
		def __init__(
		self,
		@@ -484,20 +503,9 @@ class TrafficInfluenceService:
		)
		request_metadata = operation.metadata.get("request", {}) if operation is not None else {}
		correlation_id = operation.correlation_id if operation is not None else str(uuid4())
		selected_app_instance_ids: list[UUID] = []
		if (
		traffic_influence.state == RecordTrafficInfluenceState.ACTIVE
		and traffic_influence.edge_cloud_zone_id is not None
		and self._app_instance_repo is not None
		):
		selected_app_instance_ids = await self._app_instance_repo.list_ready_ids_in_zone(
		traffic_influence.app_registration_id, traffic_influence.edge_cloud_zone_id
		selected_app_instance_ids = await self._selected_app_instance_ids(
		traffic_influence, operation, request_metadata
		)
		pinned_app_instance_id = request_metadata.get("appInstanceId")
		if pinned_app_instance_id is not None:
		selected_app_instance_ids = [
		i for i in selected_app_instance_ids if i == UUID(pinned_app_instance_id)
		]
		cloud_event = build_traffic_influence_changed_event(
		traffic_influence=traffic_influence,
		request_metadata=request_metadata,
		@@ -546,6 +554,27 @@ class TrafficInfluenceService:
		)
		)

		async def _selected_app_instance_ids(
		self,
		traffic_influence: TrafficInfluenceRecord,
		operation: Optional[Operation],
		request_metadata: dict[str, Any],
		) -> list[UUID]:
		if traffic_influence.state != RecordTrafficInfluenceState.ACTIVE:
		return []
		reported = _reported_selected_app_instance_ids(operation)
		if reported is not None:
		return reported
		if traffic_influence.edge_cloud_zone_id is None or self._app_instance_repo is None:
		return []
		selected = await self._app_instance_repo.list_ready_ids_in_zone(
		traffic_influence.app_registration_id, traffic_influence.edge_cloud_zone_id
		)
		pinned_app_instance_id = request_metadata.get("appInstanceId")
		if pinned_app_instance_id is not None:
		selected = [i for i in selected if i == UUID(pinned_app_instance_id)]
		return selected

		async def _record_skipped_delivery(
		self, callback_registration_id: UUID, operation_id: UUID, reason: str
		) -> None:

src/open_exposure_gateway/domain/edge_application_management.py

+1 −0

Changes for src/open_exposure_gateway/domain/edge_application_management.py: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -282,6 +282,7 @@ class SRMCompletedInstance(BaseModel):
		status: Literal["completed", "failed"]
		external_ref: str \| None = None
		error: dict[str, Any] \| None = None
		metadata: dict[str, Any] \| None = None

		@model_validator(mode="after")
		def _require_error_when_failed(self) -> SRMCompletedInstance:

src/open_exposure_gateway/domain/traffic_influence.py

+1 −0

Changes for src/open_exposure_gateway/domain/traffic_influence.py: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -25,6 +25,7 @@ class Subject(StrEnum):
		class NetworkCapabilityTargetApplicationReference(BaseModel):
		service_specification_ref: str
		external_app_id: str
		app_instance_id: str \| None = None


		class NetworkCapabilityTarget(BaseModel):

tests/unit/test_traffic_influence_flows.py

+149 −0

Changes for tests/unit/test_traffic_influence_flows.py: 149 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -599,6 +599,155 @@ class TestTrafficInfluenceCreateFlow:
		assert event.data["state"] == "active"
		assert "selected_appInstanceId" not in event.data

		def test_pinned_app_instance_is_forwarded_to_srm(
		self, api_client: TestClient, fake_bus: FakeDataBus
		) -> None:
		"""SRM routes to the oldest active instance in the zone unless the activate names
		one, so a client-pinned appInstanceId must reach SRM, not only its zone."""
		app_instance_id = str(uuid4())

		api_client.post(
		f"{TI_BASE}/traffic-influences", json={**TI_BODY, "appInstanceId": app_instance_id}
		)

		activate = [p for s, p in fake_bus.published if s == Subject.TASK_ACTIVATE][0]
		reference = activate["network_capability"]["target"]["application_reference"]
		assert reference["app_instance_id"] == app_instance_id

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

		activate = [p for s, p in fake_bus.published if s == Subject.TASK_ACTIVATE][0]
		reference = activate["network_capability"]["target"]["application_reference"]
		assert "app_instance_id" not in reference

		async def test_webhook_reports_the_app_instance_srm_selected(
		self,
		api_client: TestClient,
		live_ti: None,
		fake_bus: FakeDataBus,
		traffic_influence_repo: FakeTrafficInfluenceRepository,
		traffic_influence_callback_delivery_port: FakeTrafficInfluenceCallbackDeliveryPort,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		"""SRM routes to one instance and reports it; every other ready instance in the
		zone must not appear in selected_appInstanceId."""
		zone_id = UUID(TI_BODY["edgeCloudZoneId"])
		_ready_app_instance(app_instance_repo, zone_id)
		chosen = _ready_app_instance(app_instance_repo, zone_id)
		response = api_client.post(
		f"{TI_BASE}/traffic-influences",
		json={**TI_BODY, "subscriptionRequest": SUBSCRIPTION_REQUEST},
		)
		operation_id = traffic_influence_repo.rows[
		UUID(response.json()["trafficInfluenceID"])
		].operation_id

		await fake_bus.publish(
		str(Subject.OPERATION_COMPLETED),
		completion_payload(
		str(operation_id),
		instances=[
		{
		"service_instance_id": str(uuid4()),
		"zone_id": str(zone_id),
		"status": "completed",
		"metadata": {"selected_app_instance_ids": [str(chosen.app_instance_id)]},
		}
		],
		),
		)

		_, event = traffic_influence_callback_delivery_port.delivered[0]
		assert event.data["selected_appInstanceId"] == [str(chosen.app_instance_id)]

		async def test_srm_selection_survives_a_later_status_change(
		self,
		api_client: TestClient,
		live_ti: None,
		fake_bus: FakeDataBus,
		traffic_influence_repo: FakeTrafficInfluenceRepository,
		traffic_influence_callback_delivery_port: FakeTrafficInfluenceCallbackDeliveryPort,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		zone_id = UUID(TI_BODY["edgeCloudZoneId"])
		_ready_app_instance(app_instance_repo, zone_id)
		chosen_id = str(uuid4())
		response = api_client.post(
		f"{TI_BASE}/traffic-influences",
		json={**TI_BODY, "subscriptionRequest": SUBSCRIPTION_REQUEST},
		)
		operation_id = str(
		traffic_influence_repo.rows[UUID(response.json()["trafficInfluenceID"])].operation_id
		)
		await fake_bus.publish(
		str(Subject.OPERATION_COMPLETED),
		completion_payload(
		operation_id,
		instances=[
		{
		"service_instance_id": str(uuid4()),
		"zone_id": str(zone_id),
		"status": "completed",
		"metadata": {"selected_app_instance_ids": [chosen_id]},
		}
		],
		),
		)

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

		states = [e.data["state"] for _, e in traffic_influence_callback_delivery_port.delivered]
		assert states == ["active", "error", "active"]
		_, recovered = traffic_influence_callback_delivery_port.delivered[2]
		assert recovered.data["selected_appInstanceId"] == [chosen_id]

		async def test_malformed_srm_selection_falls_back_to_local_bookkeeping(
		self,
		api_client: TestClient,
		live_ti: None,
		fake_bus: FakeDataBus,
		traffic_influence_repo: FakeTrafficInfluenceRepository,
		traffic_influence_callback_delivery_port: FakeTrafficInfluenceCallbackDeliveryPort,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		zone_id = UUID(TI_BODY["edgeCloudZoneId"])
		local = _ready_app_instance(app_instance_repo, zone_id)
		response = api_client.post(
		f"{TI_BASE}/traffic-influences",
		json={**TI_BODY, "subscriptionRequest": SUBSCRIPTION_REQUEST},
		)
		operation_id = traffic_influence_repo.rows[
		UUID(response.json()["trafficInfluenceID"])
		].operation_id

		await fake_bus.publish(
		str(Subject.OPERATION_COMPLETED),
		completion_payload(
		str(operation_id),
		instances=[
		{
		"service_instance_id": str(uuid4()),
		"zone_id": str(zone_id),
		"status": "completed",
		"metadata": {"selected_app_instance_ids": ["not-a-uuid"]},
		}
		],
		),
		)

		_, event = traffic_influence_callback_delivery_port.delivered[0]
		assert event.data["state"] == "active"
		assert event.data["selected_appInstanceId"] == [str(local.app_instance_id)]

		async def test_redelivered_completion_event_delivers_webhook_once(
		self,
		api_client: TestClient,