Commit 230306ca authored by Ferran Cañellas's avatar Ferran Cañellas
Browse files

fixes broken open5gs tests 

Removes factory test (now centralized)
Renames other test files
parent 60cc0f2c

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