Merge pull request #78 from SunriseOpenOperatorPlatform/feature/add-network-open5gs (4e2739e1) · Commits · OOP / code / transformation-function-sdk

src/network/clients/open5gs/client.py

+1 −41

Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		from typing import Dict

		from pydantic import ValidationError

		from src import logger
		from src.network.core.network_interface import NetworkManagementInterface, build_flows

		from ...core import common, schemas
		from ...core import schemas

		log = logger.get_logger(__name__)

		@@ -55,44 +53,6 @@ class NetworkManager(NetworkManagementInterface):
		flow_id = flow_id_mapping[session_info.qosProfile.root]
		subscription.flowInfo = build_flows(flow_id, session_info)

		# --- Implementation of NetworkManagementInterface methods ---
		def create_qod_session(self, session_info: Dict) -> Dict:
		"""
		Creates a QoD session based on the CAMARA QoD API input.
		Maps the CAMARA QoD POST /sessions to Open5GS NEF POST /{scsAsId}/subscriptions.
		"""
		url = f"{self.base_url}/{self.scs_as_id}/subscriptions"
		# Raises ValidationError if the object is not valid.
		valid_session_info = schemas.CreateSession.model_validate(session_info)

		subscription = schemas.AsSessionWithQoSSubscription(
		notificationDestination=valid_session_info.sink,
		qosReference=valid_session_info.qosProfile,
		ueIpv4Addr=valid_session_info.device.ipv4Address,
		ueIpv6Addr=valid_session_info.device.ipv6Address,
		usageThreshold=schemas.UsageThreshold(duration=valid_session_info.duration),
		)
		self.add_core_specific_parameters(subscription)
		common.open5gs_post(url, subscription)

		def get_qod_session(self, session_id: str) -> Dict:
		"""
		Retrieves a specific Open5GS QoS Subscription details.
		Maps CAMARA QoD GET /sessions/{sessionId} to Open5GS NEF GET /
		{scsAsId}/subscriptions/{subscriptionId}.
		"""
		url = f"{self.base_url}/{self.scs_as_id}/subscriptions/{session_id}"
		common.open5gs_get(url)

		def delete_qod_session(self, session_id: str) -> None:
		"""
		Deletes a specific Open5GS QoS Subscription.
		Maps CAMARA QoD DELETE /sessions/{sessionId} to Open5GS NEF DELETE /
		{scsAsId}/subscriptions/{subscriptionId}.
		"""
		url = f"{self.base_url}/{self.scs_as_id}/subscriptions/{session_id}"
		common.open5gs_delete(url)


		# Note:
		# As this class is inheriting from NetworkManagementInterface, it is

src/network/core/network_interface.py

+1 −18

Original line number	Diff line number	Diff line
		@@ -10,7 +10,7 @@
		# - Reza Mosahebfard (reza.mosahebfard@i2cat.net)
		# - Ferran Cañellas (ferran.canellas@i2cat.net)
		##
		from abc import ABC, abstractmethod
		from abc import ABC
		from itertools import product
		from typing import Dict

		@@ -78,7 +78,6 @@ class NetworkManagementInterface(ABC):
		base_url: str
		scs_as_id: str

		@abstractmethod
		def add_core_specific_qod_parameters(
		self,
		session_info: schemas.CreateSession,
		@@ -90,7 +89,6 @@ class NetworkManagementInterface(ABC):
		"""
		pass

		@abstractmethod
		def add_core_specific_ti_parameters(
		self,
		traffic_influence_info: schemas.CreateTrafficInfluence,
		@@ -102,7 +100,6 @@ class NetworkManagementInterface(ABC):
		"""
		pass

		@abstractmethod
		def core_specific_qod_validation(self, session_info: schemas.CreateSession) -> None:
		"""
		Validates core-specific parameters for the session creation.
		@@ -117,7 +114,6 @@ class NetworkManagementInterface(ABC):
		# This method should be overridden by subclasses if needed
		pass

		@abstractmethod
		def core_specific_traffic_influence_validation(
		self, traffic_influence_info: schemas.CreateTrafficInfluence
		) -> None:
		@@ -152,7 +148,6 @@ class NetworkManagementInterface(ABC):
		return subscription

		def _build_ti_subscription(self, traffic_influence_info: Dict):

		traffic_influence_data = schemas.CreateTrafficInfluence.model_validate(
		traffic_influence_info
		)
		@@ -194,18 +189,6 @@ class NetworkManagementInterface(ABC):
		return common.as_session_with_qos_post(
		self.base_url, self.scs_as_id, subscription
		)
		valid_session_info = schemas.CreateSession.model_validate(session_info)
		self.core_specific_validation(valid_session_info)
		subscription = schemas.AsSessionWithQoSSubscription(
		notificationDestination=valid_session_info.sink,
		qosReference=valid_session_info.qosProfile,
		ueIpv4Addr=valid_session_info.device.ipv4Address,
		ueIpv6Addr=valid_session_info.device.ipv6Address,
		usageThreshold=schemas.UsageThreshold(duration=valid_session_info.duration),
		)
		self.add_core_specific_parameters(subscription)
		url = f"{self.base_url}/{self.scs_as_id}/subscriptions"
		common.as_session_with_qos_post(url, self.scs_as_id, subscription)

		def get_qod_session(self, session_id: str) -> Dict:
		"""

tests/network/test_1_factory.py

deleted100644 → 0

+0 −27

Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		import pytest

		from src.network.clients.oai.client import NetworkManager as OaiClient
		from src.network.clients.open5gs.client import NetworkManager as Open5GsClient
		from src.network.core.network_factory import NetworkClientFactory

		test_cases = [
		("open5gs", "http://192.168.124.233:30769/", "scs"),
		("oai", "http://127.0.0.1", "scs-oai"),
		]


		@pytest.mark.parametrize("client_name, base_url, scs_as_id", test_cases)
		def test_factory_network(client_name, base_url, scs_as_id):
		"""
		Test the factory pattern for the network client.
		"""
		client_class_map = {
		"open5gs": Open5GsClient,
		"oai": OaiClient,
		}
		expected_client_class = client_class_map[client_name]
		network_client = NetworkClientFactory.create_network_client(
		client_name, base_url, scs_as_id
		)
		assert isinstance(network_client, expected_client_class)

tests/network/test_2_create_qod_session.py→tests/network/test_create_qod_session.py

+24 −15

Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		import pytest

		from src.network.core.network_factory import NetworkClientFactory

		test_cases = [
		("open5gs", "http://192.168.124.233:30769/", "scs"),
		("oai", "http://127.0.0.1:8080/", "scs-oai"),
		from src.common.sdk_catalog_client import SdkCatalogClient

		OPEN5GS_TEST_CASES = [
		{
		"network": {
		"client_name": "open5gs",
		"base_url": "http://192.168.124.233:30769/",
		"scs_as_id": "scs1",
		}
		}
		]


		@pytest.mark.parametrize("client_name, base_url, scs_as_id", test_cases)
		def test_valid_input(client_name, base_url, scs_as_id):
		network_client = NetworkClientFactory.create_network_client(
		client_name, base_url, scs_as_id
		@pytest.mark.parametrize(
		"client_specs",
		OPEN5GS_TEST_CASES,
		ids=["open5gs"],
		)
		def test_valid_input_open5gs(client_specs):
		network_client = SdkCatalogClient.create_clients_from(client_specs)["network"]

		camara_session = {
		"duration": 3600,
		@@ -29,11 +36,13 @@ def test_valid_input(client_name, base_url, scs_as_id):
		network_client._build_qod_subscription(camara_session)


		@pytest.mark.parametrize("client_name, base_url, scs_as_id", test_cases)
		def test_create_qod_session(client_name, base_url, scs_as_id):
		network_client = NetworkClientFactory.create_network_client(
		client_name, base_url, scs_as_id
		@pytest.mark.parametrize(
		"client_specs",
		OPEN5GS_TEST_CASES,
		ids=["open5gs"],
		)
		def test_create_qod_session_open5gs(client_specs):
		network_client = SdkCatalogClient.create_clients_from(client_specs)["network"]

		camara_session = {
		"duration": 3600,

tests/network/test_3_create_traffic_influence.py→tests/network/test_create_traffic_influence.py

+1 −1

Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		# # -- coding: utf-8 --
		import pytest

		from src.network.core.network_factory import NetworkClientFactory

Admin message