Loading src/sunrise6g_opensdk/network/clients/open5gcore/client.py +40 −5 Original line number Diff line number Diff line # -*- coding: utf-8 -*- from pydantic import ValidationError from sunrise6g_opensdk import logger from sunrise6g_opensdk.network.core.network_interface import NetworkManagementInterface from sunrise6g_opensdk.network.core.network_interface import ( NetworkManagementInterface, build_flows, ) log = logger.get_logger(__name__) from ...core import schemas log = logger.get_logger("Open5GCore") # Usage of brand name # TODO: Define any specific parameters or methods needed for Open5GCore # In case any functionality is not implemented, raise NotImplementedError qos_support_map = { "qos-e": 1, # ToDo "qos-s": 5, "qos-m": 9, "qos-l": 9, # ToDo not yet available in Nokia RAN } class NetworkManager(NetworkManagementInterface): def __init__(self, base_url: str, scs_as_id: str): pass if not base_url: raise ValueError("base_url is required and cannot be empty.") if not scs_as_id: raise ValueError("scs_as_id is required and cannot be empty.") self.base_url = base_url self.scs_as_id = scs_as_id def core_specific_qod_validation(self, session_info: schemas.CreateSession): qos_key = session_info.qosProfile.root.strip().lower() if qos_key not in qos_support_map: supported = ", ".join(qos_support_map.keys()) raise ValidationError( f"Unsupported QoS profile '{session_info.qosProfile.root}'. " f"Supported profiles for Open5GCore are: {supported}" ) def add_core_specific_qod_parameters( self, session_info: schemas.CreateSession, subscription: schemas.AsSessionWithQoSSubscription, ) -> None: flow_id = qos_support_map[session_info.qosProfile.root] subscription.flowInfo = build_flows(flow_id, session_info) subscription.ueIpv4Addr = "192.168.6.1" # ToDo tests/common/test_invoke_network_clients.py +7 −9 Original line number Diff line number Diff line Loading @@ -18,15 +18,13 @@ NETWORK_TEST_CASES = [ "scs_as_id": "scs2", } }, # TODO: Once the functionality from QoD, Location-retrieval and # traffic influnce is validated, tests can be carried out for Open5GCore # { # "network": { # "client_name": "open5gcore", # "base_url": "http://test-open5gcore.url", # "scs_as_id": "scs3", # } # }, { "network": { "client_name": "open5gcore", "base_url": "http://test-open5gcore.url", "scs_as_id": "scs3", } }, ] Loading tests/network/test_cases.py +7 −0 Original line number Diff line number Diff line Loading @@ -14,4 +14,11 @@ test_cases = [ "scs_as_id": "scs", } }, { "network": { "client_name": "open5gcore", "base_url": "http://192.168.11.80:8080", "scs_as_id": "0001", } }, ] Loading
src/sunrise6g_opensdk/network/clients/open5gcore/client.py +40 −5 Original line number Diff line number Diff line # -*- coding: utf-8 -*- from pydantic import ValidationError from sunrise6g_opensdk import logger from sunrise6g_opensdk.network.core.network_interface import NetworkManagementInterface from sunrise6g_opensdk.network.core.network_interface import ( NetworkManagementInterface, build_flows, ) log = logger.get_logger(__name__) from ...core import schemas log = logger.get_logger("Open5GCore") # Usage of brand name # TODO: Define any specific parameters or methods needed for Open5GCore # In case any functionality is not implemented, raise NotImplementedError qos_support_map = { "qos-e": 1, # ToDo "qos-s": 5, "qos-m": 9, "qos-l": 9, # ToDo not yet available in Nokia RAN } class NetworkManager(NetworkManagementInterface): def __init__(self, base_url: str, scs_as_id: str): pass if not base_url: raise ValueError("base_url is required and cannot be empty.") if not scs_as_id: raise ValueError("scs_as_id is required and cannot be empty.") self.base_url = base_url self.scs_as_id = scs_as_id def core_specific_qod_validation(self, session_info: schemas.CreateSession): qos_key = session_info.qosProfile.root.strip().lower() if qos_key not in qos_support_map: supported = ", ".join(qos_support_map.keys()) raise ValidationError( f"Unsupported QoS profile '{session_info.qosProfile.root}'. " f"Supported profiles for Open5GCore are: {supported}" ) def add_core_specific_qod_parameters( self, session_info: schemas.CreateSession, subscription: schemas.AsSessionWithQoSSubscription, ) -> None: flow_id = qos_support_map[session_info.qosProfile.root] subscription.flowInfo = build_flows(flow_id, session_info) subscription.ueIpv4Addr = "192.168.6.1" # ToDo
tests/common/test_invoke_network_clients.py +7 −9 Original line number Diff line number Diff line Loading @@ -18,15 +18,13 @@ NETWORK_TEST_CASES = [ "scs_as_id": "scs2", } }, # TODO: Once the functionality from QoD, Location-retrieval and # traffic influnce is validated, tests can be carried out for Open5GCore # { # "network": { # "client_name": "open5gcore", # "base_url": "http://test-open5gcore.url", # "scs_as_id": "scs3", # } # }, { "network": { "client_name": "open5gcore", "base_url": "http://test-open5gcore.url", "scs_as_id": "scs3", } }, ] Loading
tests/network/test_cases.py +7 −0 Original line number Diff line number Diff line Loading @@ -14,4 +14,11 @@ test_cases = [ "scs_as_id": "scs", } }, { "network": { "client_name": "open5gcore", "base_url": "http://192.168.11.80:8080", "scs_as_id": "0001", } }, ]