Loading src/sunrise6g_opensdk/oran/adapters/i2cat_ric/ip_to_plmn_gnb_mapping.json +2 −2 Original line number Diff line number Diff line { "192.168.1.10": { "mcc": "001", "mnc": "01", "gnb_length": 28, "gnb_id": 12345, "ran_ue_id":"000008"}, "10.45.0.10": { "mcc": "214", "mnc": "07", "gnb_length": 28, "gnb_id": 67890, "ran_ue_id":"000003"} "192.168.1.10": { "mcc": "001", "mnc": "01", "gnb_length": 28, "gnb_id": 12345, "ran_ue_id":"0000000000000008"}, "10.45.0.10": { "mcc": "214", "mnc": "07", "gnb_length": 28, "gnb_id": 67890, "ran_ue_id":"0000000000000003"} } src/sunrise6g_opensdk/oran/adapters/i2cat_ric/mappings.py +2 −2 Original line number Diff line number Diff line Loading @@ -28,14 +28,14 @@ ip_to_plmn_gnb_mapping = { "mnc": "01", "gnb_length": 28, "gnb_id": 12345, "ran_ue_id": "000001", "ran_ue_id": "0000000000000001", }, "10.10.45.1": { "mcc": "214", "mnc": "07", "gnb_length": 28, "gnb_id": 67890, "ran_ue_id": "000002", "ran_ue_id": "0000000000000033", }, } Loading tests/oran/test_create_qod_session.py→tests/oran/test_create_get_delete_qod_session.py +68 −0 Original line number Diff line number Diff line Loading @@ -21,73 +21,48 @@ def id_func(val): return val["oran"]["client_name"] @pytest.mark.parametrize( "oran_client", test_cases, ids=id_func, indirect=True, ) def test_create_qod_session(oran_client: BaseOranClient): @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_create_wait_get_delete_then_missing(oran_client: BaseOranClient): """ Create a QoD policy, wait 10s, verify it exists, delete it, then verify it is no longer retrievable. """ camara_session = { "duration": 300, "device": { "ipv4Address": { "publicAddress": "10.45.0.10", "privateAddress": "10.45.0.10", } }, # Use an IP present in ip_to_plmn_gnb_mapping.json "applicationServer": {"ipv4Address": "192.168.1.10"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", # Prefer notificationDestination if present "notificationDestination": "http://localhost:40000/callback", "qosProfile": "qos-s", } response = oran_client.create_qod_session(camara_session) print(response) assert response is not None, "Response should not be None" assert isinstance(response, dict), "Response should be a dictionary" assert "policy_id" in response or "policyId" in response, "Response should contain policy id" @pytest.fixture(scope="module") def oran_policy_id(oran_client: BaseOranClient): camara_session = { "duration": 300, "device": { "ipv4Address": { "publicAddress": "10.45.0.10", "privateAddress": "10.45.0.10", } }, "applicationServer": {"ipv4Address": "192.168.1.10"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", "notificationDestination": "http://localhost:40000/callback", } # Create policy response = oran_client.create_qod_session(camara_session) oran_policy_id = response.get("policy_id") or response.get("policyId") yield oran_policy_id policy_id = response.get("policy_id") or response.get("policyId") assert policy_id, "Policy ID not returned by create_qod_session" @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_timer_wait_10_seconds(oran_client): # Wait 10 seconds time.sleep(10) @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_get_qod_session(oran_client: BaseOranClient, oran_policy_id): # Verify policy exists try: oran_client.get_qod_session(oran_policy_id) oran_client.get_qod_session(policy_id) except OranHttpError as e: pytest.fail(f"Failed to retrieve oran policy: {e}") pytest.fail(f"Policy should exist before deletion: {e}") @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_delete_qod_session(oran_client: BaseOranClient, oran_policy_id): # Delete policy try: oran_client.delete_qod_session(oran_policy_id) oran_client.delete_qod_session(policy_id) except OranHttpError as e: pytest.fail(f"Failed to delete oran policy: {e}") # Wait 10 seconds for the automated expiration in NEF time.sleep(10) # Verify deletion (expect a failure on get) with pytest.raises(OranHttpError): oran_client.get_qod_session(policy_id) tests/oran/test_create_qod_session_with_expire.py 0 → 100644 +62 −0 Original line number Diff line number Diff line # -*- coding: utf-8 -*- import time import pytest from sunrise6g_opensdk.common.sdk import Sdk as sdkclient from sunrise6g_opensdk.oran.core.base_oran_client import BaseOranClient from sunrise6g_opensdk.oran.core.common import OranHttpError from tests.oran.test_cases import test_cases @pytest.fixture(scope="module", name="oran_client") def instantiate_oran_client(request): """Fixture to create and share an ORAN client across tests""" adapter_specs = request.param adapters = sdkclient.create_adapters_from(adapter_specs) return adapters.get("oran") def id_func(val): return val["oran"]["client_name"] @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_qod_policy_lifecycle_with_expiry(oran_client: BaseOranClient): """ Create a policy with a short duration, confirm it exists, then wait for expiry and confirm it no longer exists. """ duration_seconds = 15 camara_session = { "duration": duration_seconds, "device": { "ipv4Address": { "publicAddress": "10.45.0.10", "privateAddress": "10.45.0.10", } }, "applicationServer": {"ipv4Address": "192.168.1.10"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", } # Create policy response = oran_client.create_qod_session(camara_session) policy_id = response.get("policy_id") or response.get("policyId") assert policy_id, "Policy ID not returned by create_qod_session" # Immediately check it exists try: oran_client.get_qod_session(policy_id) except OranHttpError as e: pytest.fail(f"Policy should exist right after creation: {e}") # Wait slightly longer than the duration to ensure expiry buffer_seconds = 5 time.sleep(duration_seconds + buffer_seconds) # After expiry, the policy should not exist anymore (expect OranHttpError/404) with pytest.raises(OranHttpError): oran_client.get_qod_session(policy_id) tests/oran/test_qod_notifications.py 0 → 100644 +182 −0 Original line number Diff line number Diff line # -*- coding: utf-8 -*- import json import threading import time from http.server import BaseHTTPRequestHandler, HTTPServer from socketserver import ThreadingMixIn from typing import List import pytest from sunrise6g_opensdk.common.sdk import Sdk as sdkclient from sunrise6g_opensdk.oran.core.base_oran_client import BaseOranClient from sunrise6g_opensdk.oran.core.common import OranHttpError from tests.oran.test_cases import test_cases class _ThreadingHTTPServer(ThreadingMixIn, HTTPServer): daemon_threads = True def _make_handler(storage: List[dict]): class _Handler(BaseHTTPRequestHandler): def do_POST(self): length = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(length) if length > 0 else b"" try: payload = json.loads(body.decode("utf-8") or "{}") except Exception: payload = {"raw": body.decode("utf-8", errors="ignore")} storage.append( { "path": self.path, "headers": dict(self.headers), "payload": payload, "ts": time.time(), } ) self.send_response(204) self.end_headers() def log_message(self, fmt, *args): # Silence server logs during tests return return _Handler @pytest.fixture(scope="module") def notification_server(): """Spin up a tiny HTTP server to capture ORAN NEF callbacks. Binds to 0.0.0.0:40000 so the callback URL is http://localhost:40000/callback matching existing examples in tests. """ host, port = "0.0.0.0", 40000 received: List[dict] = [] handler = _make_handler(received) httpd = _ThreadingHTTPServer((host, port), handler) thread = threading.Thread(target=httpd.serve_forever, daemon=True) thread.start() server = { "url": f"http://{host}:{port}/callback", "received": received, } try: yield server finally: httpd.shutdown() httpd.server_close() thread.join(timeout=2) @pytest.fixture(scope="module", name="oran_client") def instantiate_oran_client(request): """Fixture to create and share an ORAN client across tests""" adapter_specs = request.param adapters = sdkclient.create_adapters_from(adapter_specs) return adapters.get("oran") def id_func(val): return val["oran"]["client_name"] def _wait_for_callbacks(store: List[dict], min_new: int, timeout: float, start_len: int = 0): deadline = time.time() + timeout while time.time() < deadline: if len(store) - start_len >= min_new: return True time.sleep(0.2) return False @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_qod_expire_with_notification(oran_client: BaseOranClient, notification_server): """Create a short-lived policy and expect at least one callback, then expiry notification.""" duration_seconds = 15 camara_session = { "duration": duration_seconds, "device": { "ipv4Address": { "publicAddress": "10.45.0.10", "privateAddress": "10.45.0.10", } }, "applicationServer": {"ipv4Address": "192.168.1.10"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", "notificationDestination": notification_server["url"], } start_len = len(notification_server["received"]) response = oran_client.create_qod_session(camara_session) policy_id = response.get("policy_id") or response.get("policyId") assert policy_id, "Policy ID not returned by create_qod_session" # Expect at least one callback shortly after creation assert _wait_for_callbacks( notification_server["received"], 1, timeout=10, start_len=start_len ), "Did not receive any callback after creation" # Wait until after expiry time.sleep(duration_seconds + 5) # After expiry, policy should be gone with pytest.raises(OranHttpError): oran_client.get_qod_session(policy_id) # Expect at least one additional callback (e.g., expiry) assert _wait_for_callbacks( notification_server["received"], 2, timeout=10, start_len=start_len ), "Did not receive post-expiry callback" @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_qod_create_get_delete_with_notifications(oran_client: BaseOranClient, notification_server): """Create policy, wait, verify exists, delete, then expect deletion callback and missing on get.""" camara_session = { "duration": 300, "device": { "ipv4Address": { "publicAddress": "10.45.0.10", "privateAddress": "10.45.0.10", } }, "applicationServer": {"ipv4Address": "192.168.1.10"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", "notificationDestination": notification_server["url"], } start_len = len(notification_server["received"]) response = oran_client.create_qod_session(camara_session) policy_id = response.get("policy_id") or response.get("policyId") assert policy_id, "Policy ID not returned by create_qod_session" # Expect at least one callback shortly after creation assert _wait_for_callbacks( notification_server["received"], 1, timeout=10, start_len=start_len ), "Did not receive any callback after creation" # Wait 10 seconds and verify exists time.sleep(10) try: oran_client.get_qod_session(policy_id) except OranHttpError as e: pytest.fail(f"Policy should exist before deletion: {e}") # Delete policy oran_client.delete_qod_session(policy_id) # Expect an additional callback (e.g., deletion) assert _wait_for_callbacks( notification_server["received"], 2, timeout=10, start_len=start_len ), "Did not receive post-deletion callback" # Verify policy is gone with pytest.raises(OranHttpError): oran_client.get_qod_session(policy_id) Loading
src/sunrise6g_opensdk/oran/adapters/i2cat_ric/ip_to_plmn_gnb_mapping.json +2 −2 Original line number Diff line number Diff line { "192.168.1.10": { "mcc": "001", "mnc": "01", "gnb_length": 28, "gnb_id": 12345, "ran_ue_id":"000008"}, "10.45.0.10": { "mcc": "214", "mnc": "07", "gnb_length": 28, "gnb_id": 67890, "ran_ue_id":"000003"} "192.168.1.10": { "mcc": "001", "mnc": "01", "gnb_length": 28, "gnb_id": 12345, "ran_ue_id":"0000000000000008"}, "10.45.0.10": { "mcc": "214", "mnc": "07", "gnb_length": 28, "gnb_id": 67890, "ran_ue_id":"0000000000000003"} }
src/sunrise6g_opensdk/oran/adapters/i2cat_ric/mappings.py +2 −2 Original line number Diff line number Diff line Loading @@ -28,14 +28,14 @@ ip_to_plmn_gnb_mapping = { "mnc": "01", "gnb_length": 28, "gnb_id": 12345, "ran_ue_id": "000001", "ran_ue_id": "0000000000000001", }, "10.10.45.1": { "mcc": "214", "mnc": "07", "gnb_length": 28, "gnb_id": 67890, "ran_ue_id": "000002", "ran_ue_id": "0000000000000033", }, } Loading
tests/oran/test_create_qod_session.py→tests/oran/test_create_get_delete_qod_session.py +68 −0 Original line number Diff line number Diff line Loading @@ -21,73 +21,48 @@ def id_func(val): return val["oran"]["client_name"] @pytest.mark.parametrize( "oran_client", test_cases, ids=id_func, indirect=True, ) def test_create_qod_session(oran_client: BaseOranClient): @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_create_wait_get_delete_then_missing(oran_client: BaseOranClient): """ Create a QoD policy, wait 10s, verify it exists, delete it, then verify it is no longer retrievable. """ camara_session = { "duration": 300, "device": { "ipv4Address": { "publicAddress": "10.45.0.10", "privateAddress": "10.45.0.10", } }, # Use an IP present in ip_to_plmn_gnb_mapping.json "applicationServer": {"ipv4Address": "192.168.1.10"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", # Prefer notificationDestination if present "notificationDestination": "http://localhost:40000/callback", "qosProfile": "qos-s", } response = oran_client.create_qod_session(camara_session) print(response) assert response is not None, "Response should not be None" assert isinstance(response, dict), "Response should be a dictionary" assert "policy_id" in response or "policyId" in response, "Response should contain policy id" @pytest.fixture(scope="module") def oran_policy_id(oran_client: BaseOranClient): camara_session = { "duration": 300, "device": { "ipv4Address": { "publicAddress": "10.45.0.10", "privateAddress": "10.45.0.10", } }, "applicationServer": {"ipv4Address": "192.168.1.10"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", "notificationDestination": "http://localhost:40000/callback", } # Create policy response = oran_client.create_qod_session(camara_session) oran_policy_id = response.get("policy_id") or response.get("policyId") yield oran_policy_id policy_id = response.get("policy_id") or response.get("policyId") assert policy_id, "Policy ID not returned by create_qod_session" @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_timer_wait_10_seconds(oran_client): # Wait 10 seconds time.sleep(10) @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_get_qod_session(oran_client: BaseOranClient, oran_policy_id): # Verify policy exists try: oran_client.get_qod_session(oran_policy_id) oran_client.get_qod_session(policy_id) except OranHttpError as e: pytest.fail(f"Failed to retrieve oran policy: {e}") pytest.fail(f"Policy should exist before deletion: {e}") @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_delete_qod_session(oran_client: BaseOranClient, oran_policy_id): # Delete policy try: oran_client.delete_qod_session(oran_policy_id) oran_client.delete_qod_session(policy_id) except OranHttpError as e: pytest.fail(f"Failed to delete oran policy: {e}") # Wait 10 seconds for the automated expiration in NEF time.sleep(10) # Verify deletion (expect a failure on get) with pytest.raises(OranHttpError): oran_client.get_qod_session(policy_id)
tests/oran/test_create_qod_session_with_expire.py 0 → 100644 +62 −0 Original line number Diff line number Diff line # -*- coding: utf-8 -*- import time import pytest from sunrise6g_opensdk.common.sdk import Sdk as sdkclient from sunrise6g_opensdk.oran.core.base_oran_client import BaseOranClient from sunrise6g_opensdk.oran.core.common import OranHttpError from tests.oran.test_cases import test_cases @pytest.fixture(scope="module", name="oran_client") def instantiate_oran_client(request): """Fixture to create and share an ORAN client across tests""" adapter_specs = request.param adapters = sdkclient.create_adapters_from(adapter_specs) return adapters.get("oran") def id_func(val): return val["oran"]["client_name"] @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_qod_policy_lifecycle_with_expiry(oran_client: BaseOranClient): """ Create a policy with a short duration, confirm it exists, then wait for expiry and confirm it no longer exists. """ duration_seconds = 15 camara_session = { "duration": duration_seconds, "device": { "ipv4Address": { "publicAddress": "10.45.0.10", "privateAddress": "10.45.0.10", } }, "applicationServer": {"ipv4Address": "192.168.1.10"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", } # Create policy response = oran_client.create_qod_session(camara_session) policy_id = response.get("policy_id") or response.get("policyId") assert policy_id, "Policy ID not returned by create_qod_session" # Immediately check it exists try: oran_client.get_qod_session(policy_id) except OranHttpError as e: pytest.fail(f"Policy should exist right after creation: {e}") # Wait slightly longer than the duration to ensure expiry buffer_seconds = 5 time.sleep(duration_seconds + buffer_seconds) # After expiry, the policy should not exist anymore (expect OranHttpError/404) with pytest.raises(OranHttpError): oran_client.get_qod_session(policy_id)
tests/oran/test_qod_notifications.py 0 → 100644 +182 −0 Original line number Diff line number Diff line # -*- coding: utf-8 -*- import json import threading import time from http.server import BaseHTTPRequestHandler, HTTPServer from socketserver import ThreadingMixIn from typing import List import pytest from sunrise6g_opensdk.common.sdk import Sdk as sdkclient from sunrise6g_opensdk.oran.core.base_oran_client import BaseOranClient from sunrise6g_opensdk.oran.core.common import OranHttpError from tests.oran.test_cases import test_cases class _ThreadingHTTPServer(ThreadingMixIn, HTTPServer): daemon_threads = True def _make_handler(storage: List[dict]): class _Handler(BaseHTTPRequestHandler): def do_POST(self): length = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(length) if length > 0 else b"" try: payload = json.loads(body.decode("utf-8") or "{}") except Exception: payload = {"raw": body.decode("utf-8", errors="ignore")} storage.append( { "path": self.path, "headers": dict(self.headers), "payload": payload, "ts": time.time(), } ) self.send_response(204) self.end_headers() def log_message(self, fmt, *args): # Silence server logs during tests return return _Handler @pytest.fixture(scope="module") def notification_server(): """Spin up a tiny HTTP server to capture ORAN NEF callbacks. Binds to 0.0.0.0:40000 so the callback URL is http://localhost:40000/callback matching existing examples in tests. """ host, port = "0.0.0.0", 40000 received: List[dict] = [] handler = _make_handler(received) httpd = _ThreadingHTTPServer((host, port), handler) thread = threading.Thread(target=httpd.serve_forever, daemon=True) thread.start() server = { "url": f"http://{host}:{port}/callback", "received": received, } try: yield server finally: httpd.shutdown() httpd.server_close() thread.join(timeout=2) @pytest.fixture(scope="module", name="oran_client") def instantiate_oran_client(request): """Fixture to create and share an ORAN client across tests""" adapter_specs = request.param adapters = sdkclient.create_adapters_from(adapter_specs) return adapters.get("oran") def id_func(val): return val["oran"]["client_name"] def _wait_for_callbacks(store: List[dict], min_new: int, timeout: float, start_len: int = 0): deadline = time.time() + timeout while time.time() < deadline: if len(store) - start_len >= min_new: return True time.sleep(0.2) return False @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_qod_expire_with_notification(oran_client: BaseOranClient, notification_server): """Create a short-lived policy and expect at least one callback, then expiry notification.""" duration_seconds = 15 camara_session = { "duration": duration_seconds, "device": { "ipv4Address": { "publicAddress": "10.45.0.10", "privateAddress": "10.45.0.10", } }, "applicationServer": {"ipv4Address": "192.168.1.10"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", "notificationDestination": notification_server["url"], } start_len = len(notification_server["received"]) response = oran_client.create_qod_session(camara_session) policy_id = response.get("policy_id") or response.get("policyId") assert policy_id, "Policy ID not returned by create_qod_session" # Expect at least one callback shortly after creation assert _wait_for_callbacks( notification_server["received"], 1, timeout=10, start_len=start_len ), "Did not receive any callback after creation" # Wait until after expiry time.sleep(duration_seconds + 5) # After expiry, policy should be gone with pytest.raises(OranHttpError): oran_client.get_qod_session(policy_id) # Expect at least one additional callback (e.g., expiry) assert _wait_for_callbacks( notification_server["received"], 2, timeout=10, start_len=start_len ), "Did not receive post-expiry callback" @pytest.mark.parametrize("oran_client", test_cases, ids=id_func, indirect=True) def test_qod_create_get_delete_with_notifications(oran_client: BaseOranClient, notification_server): """Create policy, wait, verify exists, delete, then expect deletion callback and missing on get.""" camara_session = { "duration": 300, "device": { "ipv4Address": { "publicAddress": "10.45.0.10", "privateAddress": "10.45.0.10", } }, "applicationServer": {"ipv4Address": "192.168.1.10"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", "notificationDestination": notification_server["url"], } start_len = len(notification_server["received"]) response = oran_client.create_qod_session(camara_session) policy_id = response.get("policy_id") or response.get("policyId") assert policy_id, "Policy ID not returned by create_qod_session" # Expect at least one callback shortly after creation assert _wait_for_callbacks( notification_server["received"], 1, timeout=10, start_len=start_len ), "Did not receive any callback after creation" # Wait 10 seconds and verify exists time.sleep(10) try: oran_client.get_qod_session(policy_id) except OranHttpError as e: pytest.fail(f"Policy should exist before deletion: {e}") # Delete policy oran_client.delete_qod_session(policy_id) # Expect an additional callback (e.g., deletion) assert _wait_for_callbacks( notification_server["received"], 2, timeout=10, start_len=start_len ), "Did not receive post-deletion callback" # Verify policy is gone with pytest.raises(OranHttpError): oran_client.get_qod_session(policy_id)
