fix: configure OAI QoD profiles per control path (!207) · Merge requests · OOP / code / transformation-function-sdk

src/openop_tfsdk/contract/adapters/network.py

+32 −7

Changes for src/openop_tfsdk/contract/adapters/network.py: 32 added lines, 7 removed lines.
Original line number	Diff line number	Diff line
		@@ -247,11 +247,15 @@ class NetworkCapabilityAdapter:
		def __init__(self, infrastructure: Optional[dict] = None):
		infrastructure = infrastructure or {}
		self._scope = ControlPathScope(infrastructure.get("control_path_ref"))
		self._client = self.client_class(
		base_url=infrastructure.get("base_url"), scs_as_id=infrastructure.get("scs_as_id")
		)
		self._client = self.client_class(**self._client_kwargs(infrastructure))
		self._notification_url = infrastructure.get("notification_url")

		def _client_kwargs(self, infrastructure: dict) -> dict:
		return {
		"base_url": infrastructure.get("base_url"),
		"scs_as_id": infrastructure.get("scs_as_id"),
		}

		def activate(self, request: CapabilityLifecycleRequest) -> CapabilityLifecycleResult:
		self._scope.check(request)
		self._require(request.capability_kind, SupportedAction.ACTIVATE)
		@@ -405,23 +409,25 @@ class NetworkCapabilityAdapter:
		if isinstance(cause, requests.HTTPError) and cause.response is not None:
		if cause.response.status_code == 404:
		raise ResourceNotFound(f"{method}: not found") from error
		diagnostics = backend_diagnostics(error)
		log.error(
		"backend_call_failed",
		method=method,
		error_code=DomainExecutionFailed.error_code,
		error_type=type(error).__name__,
		**backend_diagnostics(error),
		**diagnostics,
		)
		raise DomainExecutionFailed(f"{method} failed") from error
		raise DomainExecutionFailed(f"{method} failed", details=diagnostics or None) from error
		except Exception as error:
		diagnostics = backend_diagnostics(error)
		log.error(
		"backend_call_failed",
		method=method,
		error_code=DomainExecutionFailed.error_code,
		error_type=type(error).__name__,
		**backend_diagnostics(error),
		**diagnostics,
		)
		raise DomainExecutionFailed(f"{method} failed") from error
		raise DomainExecutionFailed(f"{method} failed", details=diagnostics or None) from error


		class OaiCapabilityAdapter(NetworkCapabilityAdapter):
		@@ -429,6 +435,25 @@ class OaiCapabilityAdapter(NetworkCapabilityAdapter):
		client_class = OaiClient
		location_identifiers = ("network_access_id", "ipv4")

		def _client_kwargs(self, infrastructure: dict) -> dict:
		options = super()._client_kwargs(infrastructure)
		if "qos_profiles" not in infrastructure:
		return options
		profiles = infrastructure["qos_profiles"]
		message = "qos_profiles must contain nonempty, unique, valid QoS profile names"
		if (
		not isinstance(profiles, (list, tuple))
		or not profiles
		or not all(isinstance(name, str) for name in profiles)
		or len(set(profiles)) != len(profiles)
		):
		raise ValidationError(message)
		try:
		options["qos_profiles"] = tuple(schemas.QosProfileName(name).root for name in profiles)
		except pydantic.ValidationError:
		raise ValidationError(message) from None
		return options

		@staticmethod
		def capability_surface(capability_kind: str, config: Optional[dict] = None):
		return _surface(OaiClient, capability_kind)

src/openop_tfsdk/network/adapters/oai/client.py

+11 −4

Changes for src/openop_tfsdk/network/adapters/oai/client.py: 11 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -7,6 +7,7 @@
		# - Giulio Carota (giulio.carota@eurecom.fr)
		##
		from datetime import datetime, timedelta, timezone
		from typing import Optional, Sequence

		import requests

		@@ -27,7 +28,6 @@ from openop_tfsdk.network.core.schemas import (
		)

		log = logger.get_logger(__name__)
		supportedQos = ["qos-e", "qos-s", "qos-m", "qos-l"]


		class NetworkManager(BaseNetworkClient):
		@@ -38,11 +38,14 @@ class NetworkManager(BaseNetworkClient):

		capabilities = {"qod", "traffic_influence", "location_retrieval"}

		def __init__(self, base_url: str, scs_as_id: str = None):
		def __init__(
		self, base_url: str, scs_as_id: str = None, *, qos_profiles: Optional[Sequence[str]] = None
		):
		try:
		super().__init__()
		self.base_url = base_url
		self.scs_as_id = scs_as_id
		self._qos_profiles = None if qos_profiles is None else tuple(qos_profiles)
		self.identity_service_url = f"{base_url.rstrip('/')}/3gpp-ueid/v1"
		log.info(
		f"Initialized OaiNefClient with base_url: {self.base_url} and scs_as_id: {self.scs_as_id}"
		@@ -63,9 +66,13 @@ class NetworkManager(BaseNetworkClient):
		raises:
		ValidationError: If the session information does not meet core-specific requirements.
		"""
		if session_info.qosProfile.root not in supportedQos:
		if (
		self._qos_profiles is not None
		and session_info.qosProfile.root not in self._qos_profiles
		):
		raise OaiValidationError(
		f"QoS profile {session_info.qosProfile} not supported by OAI, supported profiles are {supportedQos}"
		f"QoS profile {session_info.qosProfile} not allowed by configured qos_profiles, "
		f"allowed profiles are {list(self._qos_profiles)}"
		)

		if session_info.device is None or session_info.device.ipv4Address is None:

src/openop_tfsdk/_diagnostics.py

+1 −1

Changes for src/openop_tfsdk/_diagnostics.py: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		"""Small log-only summaries; never copy backend text, payloads or input values."""
		"""Allowlisted diagnostic summaries; never copy backend text, payloads or input values."""

		from typing import get_args

tests/contract/test_network_error_details.py

0 → 100644

+86 −0

Changes for tests/contract/test_network_error_details.py: 86 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""Safe network diagnostics reach consumers through the existing error contract."""

		from types import SimpleNamespace

		import pytest
		import requests
		from structlog.testing import capture_logs

		from openop_tfsdk.contract.adapters.network import NetworkCapabilityAdapter
		from openop_tfsdk.contract.errors import DomainExecutionFailed, ResourceNotFound
		from openop_tfsdk.network.core.common import CoreHttpError


		@pytest.mark.parametrize("wrapped", [False, True])
		@pytest.mark.parametrize("status, reason", [(400, "bad_request"), (503, "service_unavailable")])
		def test_http_details_are_allowlisted_and_preserve_error_contract(wrapped, status, reason):
		marker = "PRIVATE_BACKEND_BODY_AND_URL"
		response = requests.Response()
		response.status_code = status
		response.url = "http://nef.invalid/" + marker
		response._content = marker.encode()
		error = requests.HTTPError(marker, response=response)

		def operation():
		if wrapped:
		raise CoreHttpError(marker) from error
		raise error

		adapter = NetworkCapabilityAdapter.__new__(NetworkCapabilityAdapter)
		adapter._client = SimpleNamespace(operation=operation)
		with capture_logs() as events, pytest.raises(DomainExecutionFailed) as caught:
		adapter._call("operation")
		expected = {"backend_status": status, "reason": reason}
		assert caught.value.details == expected
		assert caught.value.error_code == "DOMAIN_EXECUTION_FAILED"
		assert caught.value.retryable is False
		assert caught.value.message == "operation failed"
		assert caught.value.__cause__ is not None
		assert all(events[0][key] == value for key, value in expected.items())
		assert marker not in str(events) + str(caught.value) + str(caught.value.details)


		@pytest.mark.parametrize("wrapped", [False, True])
		@pytest.mark.parametrize(
		"error_type, reason",
		[(requests.Timeout, "timeout"), (requests.ConnectionError, "connection_failed")],
		)
		def test_transport_details_keep_only_safe_reason(wrapped, error_type, reason):
		error = error_type("PRIVATE_TRANSPORT_DESTINATION")

		def operation():
		if wrapped:
		raise CoreHttpError("PRIVATE_WRAPPER_TEXT") from error
		raise error

		adapter = NetworkCapabilityAdapter.__new__(NetworkCapabilityAdapter)
		adapter._client = SimpleNamespace(operation=operation)
		with pytest.raises(DomainExecutionFailed) as caught:
		adapter._call("operation")
		assert caught.value.details == {"reason": reason}
		assert caught.value.retryable is False


		def test_unknown_failure_does_not_invent_details():
		def operation():
		raise RuntimeError("PRIVATE_UNRECOGNIZED_TEXT")

		adapter = NetworkCapabilityAdapter.__new__(NetworkCapabilityAdapter)
		adapter._client = SimpleNamespace(operation=operation)
		with pytest.raises(DomainExecutionFailed) as caught:
		adapter._call("operation")
		assert caught.value.details is None


		def test_wrapped_http_404_remains_resource_not_found():
		response = requests.Response()
		response.status_code = 404

		def operation():
		raise CoreHttpError("private") from requests.HTTPError(response=response)

		adapter = NetworkCapabilityAdapter.__new__(NetworkCapabilityAdapter)
		adapter._client = SimpleNamespace(operation=operation)
		with pytest.raises(ResourceNotFound) as caught:
		adapter._call("operation")
		assert caught.value.details is None

tests/contract/test_network_qod.py

+1 −1

Changes for tests/contract/test_network_qod.py: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -110,7 +110,7 @@ def test_qod_behind_nat_targets_the_private_address(nef_bodies):
		@pytest.mark.parametrize(
		"kind, request_kwargs, reason",
		[
		("oai", {"profile": "QOS_PREMIUM"}, "not supported by OAI"),
		("oai", {"profile": "qos e"}, "qosProfile string_pattern_mismatch"),
		("oai", {"ipv4": None, "ipv6": "2001:db8::1"}, "OAI requires UE IPv4 Address"),
		("open5gs", {"profile": "qos-xx"}, "Open5Gs only supports"),
		("open5gs", {"profile": "qos e"}, "qosProfile string_pattern_mismatch"),

fix: configure OAI QoD profiles per control path

Merge request reports