Commit f119c345 authored by Adrian Pino's avatar Adrian Pino
Browse files

Add region & status support for get_edge_cloud_zones. Add tests

parent e8c511bd
Loading
Loading
Loading
Loading
+22 −2
Original line number Diff line number Diff line
from typing import Optional
from src.edgecloud.core.edgecloud_interface import EdgeCloudManagementInterface
from typing import Dict, List, Optional
from . import schemas
from .common import I2EdgeError, i2edge_delete, i2edge_get, i2edge_post, i2edge_post_multiform_data

class EdgeApplicationManager(EdgeCloudManagementInterface):
    def __init__(self, base_url: str):
        self.base_url = base_url

    def get_edge_cloud_zones(self) -> list[dict]:
    def get_edge_cloud_zones(self, region: Optional[str] = None, status: Optional[str] = None) -> list[dict]:
        # Note: status is not supported by i2Edge; won't be used
        try:
            params = {}
            if region is not None:
                # Use the /zone/{region} endpoint
                url = "{}/zone/{}".format(self.base_url, region)
                if status is not None:
                    params['status'] = status
                response = i2edge_get(url, params=params)
            else:
                # Use the /zones/list endpoint
                url = "{}/zones/list".format(self.base_url)
                if status is not None:
                    params['status'] = status
                response = i2edge_get(url, params=params)
            
            return response
        except I2EdgeError as e:
            raise e

        try:
            response = i2edge_get(url, params=None)
            return response
+21 −7
Original line number Diff line number Diff line
@@ -9,9 +9,9 @@ from src.edgecloud.clients.dmo.client import EdgeApplicationManager as DmoClient
# Define common test cases for all tests
test_cases = [
    ("i2edge", "http://192.168.123.237:30769/"),
    ("aeros", "http://aeros.example.com/"),
    ("piedge", "http://piedge.example.com/"),
    ("dmo", "http://dmo.example.com/")
    # ("aeros", "http://aeros.example.com/"),
    # ("piedge", "http://piedge.example.com/"),
    # ("dmo", "http://dmo.example.com/")
]

# Add an invalid client test case
@@ -37,15 +37,29 @@ def test_factory_edgecloud(client_name, base_url):


@pytest.mark.parametrize("client_name, base_url", test_cases)
def test_get_edge_cloud_zones_return_list(client_name, base_url):
def test_get_edge_cloud_zones(client_name, base_url):
    """
    Test the get_edge_cloud_zones method for each client.
    Test the format of the response from get_edge_cloud_zones for each client.
    """
    # Create the edgecloud client
    edgecloud_platform = EdgeCloudFactory.create_edgecloud_client(client_name, base_url)

    # Call the get_edge_cloud_zones function
    # Case 1: status & region (which are optional) not specified
    zones = edgecloud_platform.get_edge_cloud_zones()
    assert isinstance(zones, list), f"Expected a list of zones for {client_name}, but got {type(zones)}"
    if zones:  # Check content if the list is not empty
        assert all(isinstance(zone, dict) for zone in zones), "Each zone should be a dictionary"

    # Case 2: region specified
    zones = edgecloud_platform.get_edge_cloud_zones(region="Omega")
    assert isinstance(zones, dict), f"Expected a dict for {client_name} when region is specified, but got {type(zones)}"

    # Assert that the result is a list (or whatever the expected type is)
    # Case 3: status specified
    zones = edgecloud_platform.get_edge_cloud_zones(status="active")
    assert isinstance(zones, list), f"Expected a list of zones for {client_name}, but got {type(zones)}"
    if zones:  # Check content if the list is not empty
        assert all(isinstance(zone, dict) for zone in zones), "Each zone should be a dictionary"

    # Case 4: status & region specified
    zones = edgecloud_platform.get_edge_cloud_zones(region="Omega", status="active")
    assert isinstance(zones, dict), f"Expected a dict for {client_name} when region & status is specified, but got {type(zones)}"