implements get/delete tests for open5gs (c3957450) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/network/core/network_interface.py

+17 −1

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

		from sunrise6g_opensdk import logger
		from sunrise6g_opensdk.network.clients.errors import NetworkPlatformError
		from sunrise6g_opensdk.network.core import common, schemas

		log = logger.get_logger(__name__)
		@@ -197,9 +199,23 @@ class NetworkManagementInterface(ABC):
		dictionary containing the created session details, including its ID.
		"""
		subscription = self._build_qod_subscription(session_info)
		return common.as_session_with_qos_post(
		response = common.as_session_with_qos_post(
		self.base_url, self.scs_as_id, subscription
		)
		subscription_info: schemas.AsSessionWithQoSSubscription = (
		schemas.AsSessionWithQoSSubscription(**response)
		)
		subscription_url = subscription_info.self_.root
		subscription_id = subscription_url.split("/")[-1] if subscription_url else None
		if not subscription_id:
		log.error("Failed to retrieve QoS session ID from response")
		raise NetworkPlatformError("QoS session ID not found in response")
		session_info = schemas.SessionInfo(
		sessionId=schemas.SessionId(uuid.UUID(subscription_id)),
		qosStatus=schemas.QosStatus.REQUESTED,
		**session_info,
		)
		return session_info.model_dump()

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

src/sunrise6g_opensdk/network/core/schemas.py

+46 −0

Original line number	Diff line number	Diff line
		@@ -4,9 +4,11 @@
		# specifically focusing on the APIs needed to support CAMARA QoD.

		import ipaddress
		from datetime import datetime
		from enum import Enum
		from ipaddress import IPv4Address, IPv6Address
		from typing import Annotated
		from uuid import UUID

		from pydantic import AnyUrl, BaseModel, ConfigDict, Field, NonNegativeInt, RootModel
		from pydantic_extra_types.mac_address import MacAddress
		@@ -403,6 +405,50 @@ class CreateSession(BaseSessionInfo):
		]


		class SessionId(RootModel[UUID]):
		root: Annotated[UUID, Field(description="Session ID in UUID format")]


		class QosStatus(Enum):
		REQUESTED = "REQUESTED"
		AVAILABLE = "AVAILABLE"
		UNAVAILABLE = "UNAVAILABLE"


		class StatusInfo(Enum):
		DURATION_EXPIRED = "DURATION_EXPIRED"
		NETWORK_TERMINATED = "NETWORK_TERMINATED"
		DELETE_REQUESTED = "DELETE_REQUESTED"


		class SessionInfo(BaseSessionInfo):
		sessionId: SessionId
		duration: Annotated[
		int,
		Field(
		description='Session duration in seconds. Implementations can grant the requested session duration or set a different duration, based on network policies or conditions.\n- When `qosStatus` is "REQUESTED", the value is the duration to be scheduled, granted by the implementation.\n- When `qosStatus` is AVAILABLE", the value is the overall duration since `startedAt. When the session is extended, the value is the new overall duration of the session.\n- When `qosStatus` is "UNAVAILABLE", the value is the overall effective duration since `startedAt` until the session was terminated.\n',
		examples=[3600],
		ge=1,
		),
		]
		startedAt: Annotated[
		datetime \| None,
		Field(
		description='Date and time when the QoS status became "AVAILABLE". Not to be returned when `qosStatus` is "REQUESTED". Format must follow RFC 3339 and must indicate time zone (UTC or local).',
		examples=["2024-06-01T12:00:00Z"],
		),
		] = None
		expiresAt: Annotated[
		datetime \| None,
		Field(
		description='Date and time of the QoS session expiration. Format must follow RFC 3339 and must indicate time zone (UTC or local).\n- When `qosStatus` is "AVAILABLE", it is the limit time when the session is scheduled to finnish, if not terminated by other means.\n- When `qosStatus` is "UNAVAILABLE", it is the time when the session was terminated.\n- Not to be returned when `qosStatus` is "REQUESTED".\nWhen the session is extended, the value is the new expiration time of the session.\n',
		examples=["2024-06-01T13:00:00Z"],
		),
		] = None
		qosStatus: QosStatus
		statusInfo: StatusInfo \| None = None


		class CreateTrafficInfluence(BaseModel):
		trafficInfluenceID: str \| None = None
		apiConsumerId: str \| None = None

tests/network/test_cases.py

0 → 100644

+10 −0

Original line number	Diff line number	Diff line
		# -- coding: utf-8 --
		test_cases = [
		{
		"network": {
		"client_name": "open5gs",
		"base_url": "http://192.168.124.233:8082/",
		"scs_as_id": "scs",
		}
		}
		]

tests/network/test_create_qod_session.py

+68 −33

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.network.clients.open5gs.client import NetworkManager

		OPEN5GS_TEST_CASES = [
		{
		"network": {
		"client_name": "open5gs",
		"base_url": "http://192.168.124.233:8082/",
		"scs_as_id": "scs",
		}
		}
		]
		from sunrise6g_opensdk.network.core.common import CoreHttpError
		from sunrise6g_opensdk.network.core.network_interface import NetworkManagementInterface
		from tests.network.test_cases import test_cases


		@pytest.fixture(scope="module", name="network_client")
		def instantiate_network_client(request):
		"""Fixture to create and share a network client across tests"""
		client_specs = request.param
		clients = sdkclient.create_clients_from(client_specs)
		return clients.get("network")


		def id_func(val):
		return val["network"]["client_name"]


		@pytest.mark.parametrize(
		"client_specs",
		OPEN5GS_TEST_CASES,
		ids=["open5gs"],
		"network_client",
		test_cases,
		ids=id_func,
		indirect=True,
		)
		def test_valid_input_open5gs(client_specs):
		network_client: NetworkManager = sdkclient.create_clients_from(client_specs)[
		"network"
		]

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

		network_client._build_qod_subscription(camara_session)

		@pytest.mark.parametrize(
		"client_specs",
		OPEN5GS_TEST_CASES,
		ids=["open5gs"],
		)
		def test_create_qod_session_open5gs(client_specs):
		network_client: NetworkManager = sdkclient.create_clients_from(client_specs)[
		"network"
		]

		@pytest.fixture(scope="module")
		def qod_session_id(network_client: NetworkManagementInterface):
		camara_session = {
		"duration": 3600,
		"device": {
		"ipv4Address": {"publicAddress": "10.45.0.3", "privateAddress": "10.45.0.3"}
		"ipv4Address": {
		"publicAddress": "10.45.0.3",
		"privateAddress": "10.45.0.3",
		}
		},
		"applicationServer": {"ipv4Address": "10.45.0.1"},
		"devicePorts": {"ranges": [{"from": 0, "to": 65535}]},
		@@ -61,4 +58,42 @@ def test_create_qod_session_open5gs(client_specs):
		"qosProfile": "qos-e",
		"sink": "https://endpoint.example.com/sink",
		}
		network_client.create_qod_session(camara_session)
		try:
		response = network_client.create_qod_session(camara_session)
		assert response is not None, "Response should not be None"
		assert isinstance(response, dict), "Response should be a dictionary"
		assert "sessionId" in response, "Response should contain 'sessionId'"
		yield str(response["sessionId"])
		finally:
		pass


		@pytest.mark.parametrize(
		"network_client",
		test_cases,
		ids=id_func,
		indirect=True,
		)
		def test_create_qod_session(qod_session_id):
		assert qod_session_id is not None


		@pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True)
		def test_timer_wait_5_seconds(network_client):
		time.sleep(5)


		@pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True)
		def test_get_qod_session(network_client: NetworkManagementInterface, qod_session_id):
		try:
		network_client.get_qod_session(qod_session_id)
		except CoreHttpError as e:
		pytest.fail(f"Failed to retrieve qod session: {e}")


		@pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True)
		def test_delete_qod_session(network_client: NetworkManagementInterface, qod_session_id):
		try:
		network_client.delete_qod_session(qod_session_id)
		except CoreHttpError as e:
		pytest.fail(f"Failed to retrieve qod session: {e}")

Admin message