improving the responses to match camara style (db0f1415) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/oran/adapters/i2cat_ric/client.py

+136 −33

Original line number	Diff line number	Diff line
		@@ -7,6 +7,7 @@
		# - Miguel Catalan Cid (miguel.catalan@i2cat.net)
		##
		import json
		from datetime import datetime, timedelta, timezone
		from pathlib import Path
		from typing import Any, Dict

		@@ -222,16 +223,52 @@ class OranManager(BaseOranClient):
		return oran_common.oran_policy_post(self.base_url, self.scs_as_id, policy)
		except OranHttpError as e:
		if return_on_error:
		# Map HTTP error to CAMARA StatusInfo when returning UNAVAILABLE
		status_info = None
		if e.status_code is not None:
		if e.status_code >= 500:
		if e.status_code >= 500 or e.status_code in (408, 504):
		status_info = "NETWORK_TERMINATED"
		elif e.status_code == 410:
		status_info = "DELETE_REQUESTED"

		# Align error payload with CAMARA ErrorInfo {status, code, message}
		body = e.body if isinstance(e.body, dict) else None
		if isinstance(body, dict) and {
		"status",
		"code",
		"message",
		}.issubset(body.keys()):
		error_info = {
		"status": body.get("status"),
		"code": body.get("code"),
		"message": body.get("message"),
		}
		else:
		# Best-effort default mapping when backend doesn't provide CAMARA ErrorInfo
		code_map = {
		400: "INVALID_ARGUMENT",
		401: "UNAUTHENTICATED",
		403: "PERMISSION_DENIED",
		404: "NOT_FOUND",
		409: "CONFLICT",
		410: "GONE",
		413: "REQUEST_TOO_LARGE",
		415: "UNSUPPORTED_MEDIA_TYPE",
		422: "UNPROCESSABLE_ENTITY",
		429: "TOO_MANY_REQUESTS",
		}
		error_info = {
		"status": e.status_code,
		"code": code_map.get(e.status_code or 0, "INTERNAL_ERROR"),
		"message": (
		(body or {}).get("message") if isinstance(body, dict) else str(e)
		),
		}

		return {
		"qosStatus": "UNAVAILABLE",
		"statusInfo": status_info,
		"error": {"statusCode": e.status_code, "body": e.body},
		"error": error_info,
		}
		raise

		@@ -254,8 +291,6 @@ class OranManager(BaseOranClient):
		"applicationServerPorts": session_info.get("applicationServerPorts"),
		"qosProfile": qos_profile,
		"sink": notification_uri,
		"policy_id": policy_id,
		"policyId": policy_id,
		}

		@requires_capability("oran-qod")
		@@ -265,9 +300,11 @@ class OranManager(BaseOranClient):
		original_session: Dict \| None = None,
		fallback_unavailable: bool = False,
		) -> Dict:
		"""Retrieve an ORAN policy by ID and map to a CAMARA-like response.
		"""Return CAMARA SessionInfo-style data for a QoD session.

		Adds qosStatus=AVAILABLE and harmonizes keys while preserving original fields.
		Fetches the underlying ORAN policy to determine liveness and ID but
		intentionally shapes the response to CAMARA fields only, not leaking
		ORAN-specific attributes (policyScope, policyStatement, etc.).
		"""
		try:
		resp: Dict[str, Any] = oran_common.oran_policy_get(
		@@ -279,13 +316,10 @@ class OranManager(BaseOranClient):
		return {
		"sessionId": str(session_id),
		"qosStatus": "UNAVAILABLE",
		# Backward-compatible identifiers
		"policy_id": session_id,
		"policyId": session_id,
		}
		raise

		# Determine policy/session identifier
		# Determine policy/session identifier from ORAN response, fallback to provided session_id
		policy_id = (
		(resp or {}).get("policy_id")
		or (resp or {}).get("policyId")
		@@ -293,23 +327,33 @@ class OranManager(BaseOranClient):
		or session_id
		)

		mapped: Dict[str, Any] = dict(resp or {})
		mapped.update(
		{
		# Build a fresh CAMARA-shaped response without ORAN internals
		camara_resp: Dict[str, Any] = {
		"sessionId": str(policy_id),
		"qosStatus": "AVAILABLE",
		}
		)

		# Best-effort mapping of optional fields
		expiry_val = mapped.get("expiry")
		if isinstance(expiry_val, int) and "duration" not in mapped:
		mapped["duration"] = expiry_val
		# Always include startedAt; include expiresAt only if a duration is known
		now = datetime.now(timezone.utc)
		camara_resp["startedAt"] = now.isoformat().replace("+00:00", "Z")

		expiry_val = (resp or {}).get("expiry")
		duration_sec: int \| None = None
		if isinstance(original_session, dict) and isinstance(original_session.get("duration"), int):
		duration_sec = int(original_session.get("duration"))
		elif isinstance(expiry_val, int):
		duration_sec = int(expiry_val)
		if isinstance(duration_sec, int):
		camara_resp["expiresAt"] = (
		(now + timedelta(seconds=duration_sec)).isoformat().replace("+00:00", "Z")
		)

		if mapped.get("sink") is None and mapped.get("notificationUri"):
		mapped["sink"] = mapped.get("notificationUri")
		# sink: map ORAN notification URI to CAMARA 'sink'
		notif = (resp or {}).get("notificationUri")
		if isinstance(notif, str) and notif:
		camara_resp["sink"] = notif

		# Enrich with original requested session fields when available
		# Enrich with original requested session fields when available (CAMARA keys only)
		if isinstance(original_session, dict):
		for key in (
		"device",
		@@ -318,18 +362,77 @@ class OranManager(BaseOranClient):
		"applicationServerPorts",
		"qosProfile",
		):
		if mapped.get(key) is None and original_session.get(key) is not None:
		mapped[key] = original_session.get(key)
		if mapped.get("sink") is None and original_session.get("notificationDestination"):
		mapped["sink"] = original_session.get("notificationDestination")
		val = original_session.get(key)
		if camara_resp.get(key) is None and val is not None:
		camara_resp[key] = val
		# Map legacy key
		if camara_resp.get("sink") is None and original_session.get("notificationDestination"):
		camara_resp["sink"] = original_session.get("notificationDestination")

		# Preserve compatibility identifiers
		mapped.setdefault("policy_id", policy_id)
		mapped.setdefault("policyId", policy_id)

		return mapped
		return camara_resp

		@requires_capability("oran-qod")
		def delete_qod_session(self, session_id: str) -> None:
		"""Delete an ORAN policy by ID (maps to QoD session delete)."""
		oran_common.oran_policy_delete(self.base_url, self.scs_as_id, session_id)

		def notification_to_camara_session(
		self, notification: Dict[str, Any], original_session: Dict \| None = None
		) -> Dict[str, Any]:
		"""Translate an ORAN notification payload into a CAMARA SessionInfo-like dict.

		Example notification payload:
		{
		"info_type": "policy_ue_prb_priority",
		"subscription_id": "<uuid>",
		"data": {
		"policy_status": "ENFORCED",
		...
		}
		}

		Returns a response shaped like our GET mapping and enriches fields
		using the provided CAMARA `original_session` when available.
		"""
		session_id = (
		notification.get("subscription_id")
		or notification.get("sessionId")
		or notification.get("id")
		)
		data = notification.get("data") or {}
		policy_status = str(data.get("policy_status") or "").upper()

		# Map policy status to CAMARA qosStatus
		qos_status = "AVAILABLE" if policy_status == "ENFORCED" else "UNAVAILABLE"

		camara_resp: Dict[str, Any] = {
		"sessionId": str(session_id) if session_id is not None else None,
		"qosStatus": qos_status,
		}

		# Always include startedAt; add expiresAt only if original_session carries a duration
		now = datetime.now(timezone.utc)
		camara_resp["startedAt"] = now.isoformat().replace("+00:00", "Z")
		if isinstance(original_session, dict) and isinstance(original_session.get("duration"), int):
		duration_sec = int(original_session.get("duration"))
		camara_resp["expiresAt"] = (
		(now + timedelta(seconds=duration_sec)).isoformat().replace("+00:00", "Z")
		)

		# Enrich with original CAMARA fields if provided
		if isinstance(original_session, dict):
		for key in (
		"device",
		"applicationServer",
		"devicePorts",
		"applicationServerPorts",
		"qosProfile",
		):
		val = original_session.get(key)
		if camara_resp.get(key) is None and val is not None:
		camara_resp[key] = val
		sink = original_session.get("sink") or original_session.get("notificationDestination")
		if sink and camara_resp.get("sink") is None:
		camara_resp["sink"] = sink

		return camara_resp

tests/oran/test_create_get_delete_qod_session.py

+9 −4

Original line number	Diff line number	Diff line
		@@ -48,8 +48,10 @@ def test_create_wait_get_delete_then_missing(oran_client: BaseOranClient):
		response = oran_client.create_qod_session(camara_session)
		print("\n----- [Test] Create response -----")
		print(pformat(response))
		policy_id = response.get("policy_id") or response.get("policyId")
		assert policy_id, "Policy ID not returned by create_qod_session"
		# Save CAMARA-style session info to enrich subsequent GET mapping
		camara_session_info = dict(response)
		policy_id = response.get("sessionId") or response.get("policy_id") or response.get("policyId")
		assert policy_id, "Session ID not returned by create_qod_session"

		# Wait 10 seconds
		print("\n===== [Test] WAIT before GET =====")
		@@ -60,7 +62,8 @@ def test_create_wait_get_delete_then_missing(oran_client: BaseOranClient):
		try:
		print("\n===== [Test] GET policy =====")
		print(f"[Test] policy_id={policy_id}")
		get_resp = oran_client.get_qod_session(policy_id)
		# Pass original CAMARA session info so GET can return CAMARA-compliant data
		get_resp = oran_client.get_qod_session(policy_id, original_session=camara_session_info)
		print("[Test] GET response:")
		print(pformat(get_resp))
		except OranHttpError as e:
		@@ -70,7 +73,9 @@ def test_create_wait_get_delete_then_missing(oran_client: BaseOranClient):
		try:
		print("\n===== [Test] DELETE policy =====")
		print(f"[Test] policy_id={policy_id}")
		oran_client.delete_qod_session(policy_id)
		delete_resp = oran_client.delete_qod_session(policy_id)
		# CAMARA r3.2 specifies 204 No Content for delete
		assert delete_resp is None
		except OranHttpError as e:
		pytest.fail(f"Failed to delete oran policy: {e}")

tests/oran/test_create_qod_session_with_expire.py

+7 −6

Original line number	Diff line number	Diff line
		@@ -50,14 +50,15 @@ def test_qod_policy_lifecycle_with_expiry(oran_client: BaseOranClient):
		response = oran_client.create_qod_session(camara_session)
		print("\n----- [Test] Create response -----")
		print(pformat(response))
		policy_id = response.get("policy_id") or response.get("policyId")
		assert policy_id, "Policy ID not returned by create_qod_session"
		session_id = response.get("sessionId")
		assert session_id, "Session ID not returned by create_qod_session"

		# Immediately check it exists
		try:
		print("\n===== [Test] GET after create =====")
		get_resp = oran_client.get_qod_session(policy_id)
		print(f"[Test] policy_id={policy_id}")
		# Provide original CAMARA response to enrich GET mapping
		get_resp = oran_client.get_qod_session(session_id, original_session=dict(response))
		print(f"[Test] policy_id={session_id}")
		print(pformat(get_resp))
		except OranHttpError as e:
		pytest.fail(f"Policy should exist right after creation: {e}")
		@@ -71,7 +72,7 @@ def test_qod_policy_lifecycle_with_expiry(oran_client: BaseOranClient):

		# After expiry, the policy should not exist anymore (expect OranHttpError/404)
		print("\n===== [Test] GET after expiry (expect error) =====")
		print(f"[Test] policy_id={policy_id}")
		print(f"[Test] session_id={session_id}")
		with pytest.raises(OranHttpError) as excinfo:
		oran_client.get_qod_session(policy_id)
		oran_client.get_qod_session(session_id)
		print(f"[Test] Received expected error: {excinfo.value}")

tests/oran/test_qod_notifications.py

+14 −12

Original line number	Diff line number	Diff line
		@@ -128,8 +128,8 @@ def test_qod_expire_with_notification(oran_client: BaseOranClient, notification_
		response = oran_client.create_qod_session(camara_session)
		print("\n----- [Test] Create response -----")
		print(pformat(response))
		policy_id = response.get("policy_id") or response.get("policyId")
		assert policy_id, "Policy ID not returned by create_qod_session"
		session_id = response.get("sessionId")
		assert session_id, "Session ID not returned by create_qod_session"

		# Expect at least one callback shortly after creation
		print("\n===== [Test] WAIT for creation callback =====")
		@@ -149,9 +149,9 @@ def test_qod_expire_with_notification(oran_client: BaseOranClient, notification_

		# After expiry, policy should be gone
		print("\n===== [Test] GET after expiry (expect error) =====")
		print(f"[Test] policy_id={policy_id}")
		print(f"[Test] session_id={session_id}")
		with pytest.raises(OranHttpError):
		oran_client.get_qod_session(policy_id)
		oran_client.get_qod_session(session_id)

		# Expect at least one additional callback (e.g., expiry)
		print("\n===== [Test] WAIT for expiry callback =====")
		@@ -189,8 +189,9 @@ def test_qod_create_get_delete_with_notifications(oran_client: BaseOranClient, n
		response = oran_client.create_qod_session(camara_session)
		print("\n----- [Test] Create response -----")
		print(pformat(response))
		policy_id = response.get("policy_id") or response.get("policyId")
		assert policy_id, "Policy ID not returned by create_qod_session"
		camara_session_info2 = dict(response)
		session_id2 = response.get("sessionId")
		assert session_id2, "Session ID not returned by create_qod_session"

		# Expect at least one callback shortly after creation
		print("\n===== [Test] WAIT for creation callback =====")
		@@ -207,9 +208,9 @@ def test_qod_create_get_delete_with_notifications(oran_client: BaseOranClient, n
		print("[Test] Sleeping 10s")
		time.sleep(10)
		print("\n===== [Test] GET policy =====")
		print(f"[Test] policy_id={policy_id}")
		print(f"[Test] session_id={session_id2}")
		try:
		get_resp = oran_client.get_qod_session(policy_id)
		get_resp = oran_client.get_qod_session(session_id2, original_session=camara_session_info2)
		print("\n----- [Test] GET response -----")
		print(pformat(get_resp))
		except OranHttpError as e:
		@@ -217,8 +218,9 @@ def test_qod_create_get_delete_with_notifications(oran_client: BaseOranClient, n

		# Delete policy
		print("\n===== [Test] DELETE policy =====")
		print(f"[Test] policy_id={policy_id}")
		oran_client.delete_qod_session(policy_id)
		print(f"[Test] session_id={session_id2}")
		delete_resp = oran_client.delete_qod_session(session_id2)
		assert delete_resp is None

		# Give time for deletion callback to arrive
		print("\n===== [Test] WAIT for deletion callback =====")
		@@ -236,6 +238,6 @@ def test_qod_create_get_delete_with_notifications(oran_client: BaseOranClient, n

		# Verify policy is gone
		print("\n===== [Test] GET after DELETE (expect error) =====")
		print(f"[Test] policy_id={policy_id}")
		print(f"[Test] session_id={session_id2}")
		with pytest.raises(OranHttpError):
		oran_client.get_qod_session(policy_id)
		oran_client.get_qod_session(session_id2)

Admin message