Merge pull request #98 from SunriseOpenOperatorPlatform/fix/harmonize-tests (08c9bea4) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/network/core/network_interface.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -291,7 +291,7 @@ class NetworkManagementInterface(ABC):
		self.base_url, self.scs_as_id, resource_id, subscription
		)

		traffic_influence_info.trafficInfluenceID = resource_id
		traffic_influence_info["trafficInfluenceID"] = resource_id
		return traffic_influence_info

		def delete_traffic_influence_resource(self, resource_id: str) -> None:

src/sunrise6g_opensdk/network/core/schemas.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -157,6 +157,8 @@ class AsSessionWithQoSSubscription(BaseModel):
		ueIpv4Addr: ipaddress.IPv4Address \| None = None
		ueIpv6Addr: ipaddress.IPv6Address \| None = None
		macAddr: MacAddress \| None = None
		snssai: Snssai \| None = None
		dnn: str \| None = None
		usageThreshold: UsageThreshold \| None = None
		sponsorInfo: SponsorInformation \| None = None
		qosMonInfo: QosMonitoringInformationModel \| None = None

tests/network/test_cases.py

+8 −1

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

tests/network/test_create_traffic_influence.py

+82 −8

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.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")

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

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

		@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 = sdkclient.create_network_client(client_name, base_url, scs_as_id)

		@pytest.mark.parametrize(
		"network_client",
		test_cases,
		ids=id_func,
		indirect=True,
		)
		def test_valid_input(network_client: NetworkManagementInterface):

		ti_session = {
		"device": {
		@@ -22,9 +41,8 @@ def test_valid_input(client_name, base_url, scs_as_id):
		network_client._build_ti_subscription(ti_session)


		@pytest.mark.parametrize("client_name, base_url, scs_as_id", test_cases)
		def test_create_traffic_influence(client_name, base_url, scs_as_id):
		network_client = sdkclient.create_network_client(client_name, base_url, scs_as_id)
		@pytest.fixture(scope="module")
		def traffic_influence_id(network_client: NetworkManagementInterface):

		ti_session = {
		"device": {
		@@ -35,4 +53,60 @@ def test_create_traffic_influence(client_name, base_url, scs_as_id):
		"appInstanceId": "172.21.18.3",
		"notificationUri": "https://endpoint.example.com/sink",
		}
		network_client.create_traffic_influence_resource(ti_session)
		try:
		response = network_client.create_traffic_influence_resource(ti_session)
		assert response is not None, "Response should not be None"
		assert isinstance(response, dict), "Response should be a dictionary"
		assert (
		"trafficInfluenceID" in response
		), "Response should contain 'trafficInfluenceID'"
		yield str(response["trafficInfluenceID"])
		finally:
		pass


		@pytest.mark.parametrize(
		"network_client",
		test_cases,
		ids=id_func,
		indirect=True,
		)
		def test_create_traffic_influence(traffic_influence_id):
		assert traffic_influence_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_put_traffic_influence_session(
		network_client: NetworkManagementInterface, traffic_influence_id
		):
		try:
		ti_session = {
		"device": {
		"ipv4Address": {
		"publicAddress": "12.1.2.31",
		"privateAddress": "12.1.2.31",
		}
		},
		"edgeCloudZoneId": "edge",
		"appId": "testSdk-ffff-aaaa-c0ffe",
		"appInstanceId": "172.21.18.5",
		"notificationUri": "https://endpoint.example.com/sink",
		}
		network_client.put_traffic_influence_resource(traffic_influence_id, ti_session)
		except CoreHttpError as e:
		pytest.fail(f"Failed to update traffic influence session: {e}")


		@pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True)
		def test_delete_traffic_influence_session(
		network_client: NetworkManagementInterface, traffic_influence_id
		):
		try:
		network_client.delete_traffic_influence_resource(traffic_influence_id)
		except CoreHttpError as e:
		pytest.fail(f"Failed to delete traffic influence: {e}")

Original line number	Diff line number	Diff line
		@@ -157,6 +157,8 @@ class AsSessionWithQoSSubscription(BaseModel):
		ueIpv4Addr: ipaddress.IPv4Address \| None = None
		ueIpv6Addr: ipaddress.IPv6Address \| None = None
		macAddr: MacAddress \| None = None
		snssai: Snssai \| None = None
		dnn: str \| None = None
		usageThreshold: UsageThreshold \| None = None
		sponsorInfo: SponsorInformation \| None = None
		qosMonInfo: QosMonitoringInformationModel \| None = None

Admin message