Loading src/network/clients/open5gs/client.py +3 −2 Changes for src/network/clients/open5gs/client.py: 3 added lines, 2 removed lines. Original line number Diff line number Diff line Loading @@ -2,10 +2,11 @@ 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 from ...core import schemas from ...core import common, schemas log = logger.get_logger(__name__) Loading src/network/core/common.py +1 −1 Changes for src/network/core/common.py: 1 added line, 1 removed line. Original line number Diff line number Diff line # -*- coding: utf-8 -*- # Common utilities (errors, HTTP helpers) used by the Open5GS client implementation (client.py). from pydantic import BaseModel import requests from pydantic import BaseModel from src import logger Loading src/network/core/schemas.py +1 −1 Changes for src/network/core/schemas.py: 1 added line, 1 removed line. Original line number Diff line number Diff line Loading @@ -5,11 +5,11 @@ import ipaddress from enum import Enum from ipaddress import IPv4Address, IPv6Address from typing import Annotated from pydantic import AnyUrl, BaseModel, ConfigDict, Field, NonNegativeInt, RootModel from pydantic_extra_types.mac_address import MacAddress from ipaddress import IPv4Address, IPv6Address class FlowDirection(Enum): Loading tests/network/test_1_factory copy.py 0 → 100644 +24 −0 Changes for tests/network/test_1_factory copy.py: 24 added lines, 0 removed lines. Original line number Diff line number Diff line # -*- coding: utf-8 -*- import pytest 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"), ] @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, } 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 0 → 100644 +48 −0 Changes for tests/network/test_2_create_qod_session.py: 48 added lines, 0 removed lines. 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"), ] @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 ) camara_session = { "duration": 3600, "device": { "ipv4Address": {"publicAddress": "10.45.0.3", "privateAddress": "10.45.0.3"} }, "applicationServer": {"ipv4Address": "10.45.0.1"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", "sink": "https://endpoint.example.com/sink", } network_client._build_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 ) camara_session = { "duration": 3600, "device": { "ipv4Address": {"publicAddress": "10.45.0.3", "privateAddress": "10.45.0.3"} }, "applicationServer": {"ipv4Address": "10.45.0.1"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", "sink": "https://endpoint.example.com/sink", } network_client.create_qod_session(camara_session) Loading
src/network/clients/open5gs/client.py +3 −2 Changes for src/network/clients/open5gs/client.py: 3 added lines, 2 removed lines. Original line number Diff line number Diff line Loading @@ -2,10 +2,11 @@ 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 from ...core import schemas from ...core import common, schemas log = logger.get_logger(__name__) Loading
src/network/core/common.py +1 −1 Changes for src/network/core/common.py: 1 added line, 1 removed line. Original line number Diff line number Diff line # -*- coding: utf-8 -*- # Common utilities (errors, HTTP helpers) used by the Open5GS client implementation (client.py). from pydantic import BaseModel import requests from pydantic import BaseModel from src import logger Loading
src/network/core/schemas.py +1 −1 Changes for src/network/core/schemas.py: 1 added line, 1 removed line. Original line number Diff line number Diff line Loading @@ -5,11 +5,11 @@ import ipaddress from enum import Enum from ipaddress import IPv4Address, IPv6Address from typing import Annotated from pydantic import AnyUrl, BaseModel, ConfigDict, Field, NonNegativeInt, RootModel from pydantic_extra_types.mac_address import MacAddress from ipaddress import IPv4Address, IPv6Address class FlowDirection(Enum): Loading
tests/network/test_1_factory copy.py 0 → 100644 +24 −0 Changes for tests/network/test_1_factory copy.py: 24 added lines, 0 removed lines. Original line number Diff line number Diff line # -*- coding: utf-8 -*- import pytest 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"), ] @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, } 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 0 → 100644 +48 −0 Changes for tests/network/test_2_create_qod_session.py: 48 added lines, 0 removed lines. 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"), ] @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 ) camara_session = { "duration": 3600, "device": { "ipv4Address": {"publicAddress": "10.45.0.3", "privateAddress": "10.45.0.3"} }, "applicationServer": {"ipv4Address": "10.45.0.1"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", "sink": "https://endpoint.example.com/sink", } network_client._build_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 ) camara_session = { "duration": 3600, "device": { "ipv4Address": {"publicAddress": "10.45.0.3", "privateAddress": "10.45.0.3"} }, "applicationServer": {"ipv4Address": "10.45.0.1"}, "devicePorts": {"ranges": [{"from": 0, "to": 65535}]}, "applicationServerPorts": {"ranges": [{"from": 0, "to": 65535}]}, "qosProfile": "qos-e", "sink": "https://endpoint.example.com/sink", } network_client.create_qod_session(camara_session)