More open5gs fixes (d245b603) · Commits · OOP / code / transformation-function-sdk

src/network/clients/open5gs/client.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -38,13 +38,13 @@ class NetworkManager(NetworkManagementInterface):
		log.error(f"Failed to initialize Open5GSClient: {e}")
		raise e

		def core_specific_validation(self, session_info: schemas.CreateSession):
		def core_specific_qod_validation(self, session_info: schemas.CreateSession):
		if session_info.qosProfile.root not in flow_id_mapping.keys():
		raise ValidationError(
		f"Open5Gs only supports these qos-profiles: {', '.join(flow_id_mapping.keys())}"
		)

		def add_core_specific_parameters(
		def add_core_specific_qod_parameters(
		self,
		session_info: schemas.CreateSession,
		subscription: schemas.AsSessionWithQoSSubscription,

src/network/core/common.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -35,7 +35,7 @@ def _make_request(method: str, url: str, data=None):
		def as_session_with_qos_post(
		base_url: str, scs_as_id: str, model_payload: BaseModel
		) -> dict:
		data = model_payload.model_dump_json(exclude_none=True)
		data = model_payload.model_dump_json(exclude_none=True, by_alias=True)
		url = as_session_with_qos_build_url(base_url, scs_as_id)
		return _make_request("POST", url, data=data)

src/network/core/network_interface.py

+16 −5

Original line number	Diff line number	Diff line
		@@ -29,7 +29,9 @@ def flatten_port_spec(ports_spec: schemas.PortsSpec \| None) -> list[str]:
		flat_ports.extend([str(port) for port in ports_spec.ports])
		if ports_spec and ports_spec.ranges:
		has_ranges = True
		flat_ports.extend([f"{range.from_}-{range.to}" for range in ports_spec.ranges])
		flat_ports.extend(
		[f"{range.from_.root}-{range.to.root}" for range in ports_spec.ranges]
		)
		if not has_ports and not has_ranges:
		flat_ports.append("0-65535")
		return flat_ports
		@@ -43,19 +45,26 @@ def build_flows(
		server_ports = flatten_port_spec(session_info.applicationServerPorts)
		ports_combis = list(product(device_ports, server_ports))

		device_ip = session_info.device.ipv4Address or session_info.device.ipv4Address
		device_ip = session_info.device.ipv4Address or session_info.device.ipv6Address
		if isinstance(device_ip, schemas.DeviceIpv6Address):
		device_ip = device_ip.root
		else: # IPv4
		device_ip = (
		device_ip.root.publicAddress.root or device_ip.root.privateAddress.root
		)
		device_ip = str(device_ip)
		server_ip = (
		session_info.applicationServer.ipv4Address
		or session_info.applicationServer.ipv6Address
		)

		server_ip = server_ip.root
		flow_descrs = []
		for device_port, server_port in ports_combis:
		flow_descrs.append(
		f"permit in ip from {device_ip} {device_port} to {server_ip} {server_port}"
		)
		flow_descrs.append(
		f"permit out ip from {device_ip} {device_port} to {server_ip} {server_port}"
		f"permit out ip from {server_ip} {server_port} to {device_ip} {device_port}"
		)
		flows = [
		schemas.FlowInfo(flowId=flow_id, flowDescriptions=[", ".join(flow_descrs)])
		@@ -130,7 +139,9 @@ class NetworkManagementInterface(ABC):
		# This method should be overridden by subclasses if needed
		pass

		def _build_qod_subscription(self, session_info: Dict) -> None:
		def _build_qod_subscription(
		self, session_info: Dict
		) -> schemas.AsSessionWithQoSSubscription:
		valid_session_info = schemas.CreateSession.model_validate(session_info)
		device_ipv4 = None
		if valid_session_info.device.ipv4Address:

tests/network/test_create_qod_session.py

+11 −5

Original line number	Diff line number	Diff line
		@@ -2,13 +2,14 @@
		import pytest

		from src.common.sdk_catalog_client import SdkCatalogClient
		from src.network.clients.open5gs.client import NetworkManager

		OPEN5GS_TEST_CASES = [
		{
		"network": {
		"client_name": "open5gs",
		"base_url": "http://192.168.124.233:30769/",
		"scs_as_id": "scs1",
		"base_url": "http://192.168.124.233:8082/",
		"scs_as_id": "scs",
		}
		}
		]
		@@ -20,7 +21,9 @@ OPEN5GS_TEST_CASES = [
		ids=["open5gs"],
		)
		def test_valid_input_open5gs(client_specs):
		network_client = SdkCatalogClient.create_clients_from(client_specs)["network"]
		network_client: NetworkManager = SdkCatalogClient.create_clients_from(client_specs)[
		"network"
		]

		camara_session = {
		"duration": 3600,
		@@ -33,7 +36,8 @@ def test_valid_input_open5gs(client_specs):
		"qosProfile": "qos-e",
		"sink": "https://endpoint.example.com/sink",
		}
		network_client._build_qod_subscription(camara_session)
		subscription = network_client._build_qod_subscription(camara_session)
		print(subscription.model_dump_json(exclude_none=True, by_alias=True))


		@pytest.mark.parametrize(
		@@ -42,7 +46,9 @@ def test_valid_input_open5gs(client_specs):
		ids=["open5gs"],
		)
		def test_create_qod_session_open5gs(client_specs):
		network_client = SdkCatalogClient.create_clients_from(client_specs)["network"]
		network_client: NetworkManager = SdkCatalogClient.create_clients_from(client_specs)[
		"network"
		]

		camara_session = {
		"duration": 3600,

Admin message