diff --git a/src/sunrise6g_opensdk/edgecloud/adapters/i2edge/client.py b/src/sunrise6g_opensdk/edgecloud/adapters/i2edge/client.py index 336fc3463a90e3331642c5ae1629a38e67f68882..70fa247f0f36911f76ce8460fffc7a71dbbffd7e 100644 --- a/src/sunrise6g_opensdk/edgecloud/adapters/i2edge/client.py +++ b/src/sunrise6g_opensdk/edgecloud/adapters/i2edge/client.py @@ -15,8 +15,7 @@ from pydantic import ValidationError from requests import Response from sunrise6g_opensdk import logger -from sunrise6g_opensdk.edgecloud.core import gsma_schemas -from sunrise6g_opensdk.edgecloud.core import schemas as camara_schemas +from sunrise6g_opensdk.edgecloud.core import camara_schemas, gsma_schemas from sunrise6g_opensdk.edgecloud.core.edgecloud_interface import ( EdgeCloudManagementInterface, ) diff --git a/src/sunrise6g_opensdk/edgecloud/adapters/kubernetes/client.py b/src/sunrise6g_opensdk/edgecloud/adapters/kubernetes/client.py index a02d85d4e0e4725ffc7cd90ca93a733a819c5233..c315851d1ee3bcf80ff24213ce4391409a1c771f 100644 --- a/src/sunrise6g_opensdk/edgecloud/adapters/kubernetes/client.py +++ b/src/sunrise6g_opensdk/edgecloud/adapters/kubernetes/client.py @@ -23,7 +23,7 @@ from sunrise6g_opensdk.edgecloud.adapters.kubernetes.lib.utils.connector_db impo from sunrise6g_opensdk.edgecloud.adapters.kubernetes.lib.utils.kubernetes_connector import ( KubernetesConnector, ) -from sunrise6g_opensdk.edgecloud.core import schemas as camara_schemas +from sunrise6g_opensdk.edgecloud.core import camara_schemas from sunrise6g_opensdk.edgecloud.core.edgecloud_interface import ( EdgeCloudManagementInterface, ) diff --git a/src/sunrise6g_opensdk/edgecloud/core/schemas.py b/src/sunrise6g_opensdk/edgecloud/core/camara_schemas.py similarity index 100% rename from src/sunrise6g_opensdk/edgecloud/core/schemas.py rename to src/sunrise6g_opensdk/edgecloud/core/camara_schemas.py diff --git a/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py b/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py index c2ac41afb4cef4d1736e1e2e2c04fa3eabac1b0e..1a1a661fa081c50a607fc968b5e6f243883b4451 100644 --- a/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py +++ b/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py @@ -64,7 +64,7 @@ class OSType(BaseModel): class Flavour(BaseModel): flavourId: str cpuArchType: Literal["ISA_X86", "ISA_X86_64", "ISA_ARM_64"] - supportedOSTypes: List[OSType] = Field(..., min_items=1) + supportedOSTypes: List[OSType] = Field(..., min_length=1) numCPU: int memorySize: int storageSize: int @@ -105,9 +105,9 @@ class ZoneServiceLevelObjsInfo(BaseModel): class ZoneRegisteredData(BaseModel): zoneId: str - reservedComputeResources: List[ComputeResourceInfo] = Field(..., min_items=1) - computeResourceQuotaLimits: List[ComputeResourceInfo] = Field(..., min_items=1) - flavoursSupported: List[Flavour] = Field(..., min_items=1) + reservedComputeResources: List[ComputeResourceInfo] = Field(..., min_length=1) + computeResourceQuotaLimits: List[ComputeResourceInfo] = Field(..., min_length=1) + flavoursSupported: List[Flavour] = Field(..., min_length=1) networkResources: Optional[NetworkResources] = None zoneServiceLevelObjsInfo: Optional[ZoneServiceLevelObjsInfo] = None @@ -146,6 +146,8 @@ class Artefact(BaseModel): # ApplicationOnboardingManagement # --------------------------- +# Responses + class AppDeploymentZone(BaseModel): countryCode: str @@ -202,10 +204,26 @@ class ApplicationModel(BaseModel): onboardStatusInfo: Literal["PENDING", "ONBOARDED", "DEBOARDING", "REMOVED", "FAILED"] +# Entry + + +class AppOnboardManifestGSMA(BaseModel): + appId: str + appProviderId: str + appDeploymentZones: List[str] + appMetaData: AppMetaData + appQoSProfile: AppQoSProfile + appComponentSpecs: List[AppComponentSpec] + appStatusCallbackLink: str + edgeAppFQDN: str + + # --------------------------- # ApplicationDeploymentManagement # --------------------------- +# Responses + class AppInstance(BaseModel): zoneId: str @@ -224,3 +242,47 @@ class ZoneIdentifier(BaseModel): class ZoneIdentifierList(RootModel[List[ZoneIdentifier]]): pass + + +# Entry + + +class ZoneInfo(BaseModel): + zoneId: str + flavourId: str + resourceConsumption: str + resPool: str + + +class AppDeployPayloadGSMA(BaseModel): + appId: str + appVersion: str + appProviderId: str + zoneInfo: ZoneInfo + appInstCallbackLink: str + + +# --------------------------- +# ApplicationUpdateManagement +# --------------------------- + + +class AppUpdQoSProfile(BaseModel): + latencyConstraints: str + bandwidthRequired: int + mobilitySupport: bool + multiUserClients: str + noOfUsersPerAppInst: int + appProvisioning: bool + + +class PatchAppComponentSpec(BaseModel): + serviceNameNB: str + serviceNameEW: str + componentName: str + artefactId: str + + +class PatchOnboardedAppGSMA(BaseModel): + appUpdQoSProfile: AppUpdQoSProfile + appComponentSpecs: List[PatchAppComponentSpec] diff --git a/tests/edgecloud/test_config.py b/tests/edgecloud/test_config_camara.py similarity index 95% rename from tests/edgecloud/test_config.py rename to tests/edgecloud/test_config_camara.py index 156ff46ca8a08277031c53a4018eb59d3c3dcd39..6ce39c296215859b52f094cf9dd6087e5c1d4514 100644 --- a/tests/edgecloud/test_config.py +++ b/tests/edgecloud/test_config_camara.py @@ -41,7 +41,7 @@ CONFIG = { "componentName": "my-component", "networkInterfaces": [ { - "interfaceId": "c3e1e6d4-1a5c-4a3b-9e4a-5f3d7b6c8e01", + "interfaceId": "eth0", "protocol": "TCP", "port": 8080, "visibilityType": "VISIBILITY_EXTERNAL", @@ -102,7 +102,7 @@ CONFIG = { "componentName": "aeros-component", "networkInterfaces": [ { - "interfaceId": "", + "interfaceId": "eth0", "protocol": "TCP", "port": 9090, "visibilityType": "VISIBILITY_INTERNAL", @@ -163,13 +163,13 @@ CONFIG = { { "protocol": "TCP", "port": 80, - "interfaceId": "8f4b02a9-73ba-4fde-9d93-3b1a6e7c5d9f", + "interfaceId": "http_interface", "visibilityType": "VISIBILITY_EXTERNAL", }, { "protocol": "TCP", "port": 443, - "interfaceId": "e5c9a2b1-3d7f-4b8e-a6c4-1f9d8b7a6c0e", + "interfaceId": "https_interface", "visibilityType": "VISIBILITY_EXTERNAL", }, ], diff --git a/tests/edgecloud/test_config_gsma.py b/tests/edgecloud/test_config_gsma.py new file mode 100644 index 0000000000000000000000000000000000000000..94fea67121eaaffe57b0ac6b87352b78b37b5b6b --- /dev/null +++ b/tests/edgecloud/test_config_gsma.py @@ -0,0 +1,90 @@ +CONFIG = { + "i2edge": { + "ZONE_ID": "f0662bfe-1d90-5f59-a759-c755b3b69b93", + "ARTEFACT_ID": "9c9143f0-f44f-49df-939e-1e8b891ba8f5", + "ARTEFACT_NAME": "i2edgechart", + "REPO_NAME": "github-cesar", + "REPO_TYPE": "PUBLICREPO", + "REPO_URL": "https://cesarcajas.github.io/helm-charts-examples/", + "APP_ONBOARD_MANIFEST_GSMA": { + "appId": "demo-app-id", + "appProviderId": "Y89TSlxMPDKlXZz7rN6vU2y", + "appDeploymentZones": [ + "Dmgoc-y2zv97lar0UKqQd53aS6MCTTdoGMY193yvRBYgI07zOAIktN2b9QB2THbl5Gqvbj5Zp92vmNeg7v4M" + ], + "appMetaData": { + "appName": "pj1iEkprop", + "version": "string", + "appDescription": "stringstringstri", + "mobilitySupport": False, + "accessToken": "MfxADOjxDgBhMrqmBeG8XdQFLp2XviG3cZ_LM7uQKc9b", + "category": "IOT", + }, + "appQoSProfile": { + "latencyConstraints": "NONE", + "bandwidthRequired": 1, + "multiUserClients": "APP_TYPE_SINGLE_USER", + "noOfUsersPerAppInst": 1, + "appProvisioning": True, + }, + "appComponentSpecs": [ + { + "serviceNameNB": "k8yyElSyJN4ctbNVqwodEQNUoGb2EzOEt4vQBjGnPii_5", + "serviceNameEW": "iDm08OZN", + "componentName": "HIEWqstajCmZJQmSFUj0kNHZ0xYvKWq720BKt8wjA41p", + "artefactId": "9c9143f0-f44f-49df-939e-1e8b891ba8f5", + } + ], + "appStatusCallbackLink": "string", + "edgeAppFQDN": "string", + }, + "APP_DEPLOY_PAYLOAD_GSMA": { + "appId": "demo-app-id", + "appVersion": "string", + "appProviderId": "Y89TSlxMPDKlXZz7rN6vU2y", + "zoneInfo": { + "zoneId": "f0662bfe-1d90-5f59-a759-c755b3b69b93", + "flavourId": "6881e358535a2eaedcb27214", + "resourceConsumption": "RESERVED_RES_AVOID", + "resPool": "ySIT0LuZ6ApHs0wlyGZve", + }, + "appInstCallbackLink": "string", + }, + "PATCH_ONBOARDED_APP_GSMA": { + "appUpdQoSProfile": { + "latencyConstraints": "NONE", + "bandwidthRequired": 1, + "mobilitySupport": False, + "multiUserClients": "APP_TYPE_SINGLE_USER", + "noOfUsersPerAppInst": 1, + "appProvisioning": True, + }, + "appComponentSpecs": [ + { + "serviceNameNB": "7CI_9d4lAK90vU4ASUkKxYdQjsv3y3IuwucISSQ6lG5_EMqeyVUHPIhwa5", + "serviceNameEW": "tPihoUFj30938Bu9blpsHkvsec1iA7gqZZRMpsx6o7aSSj5", + "componentName": "YCAhqPadfld8y68wJfTc6QNGguI41z", + "artefactId": "9c9143f0-f44f-49df-939e-1e8b891ba8f5", + }, + { + "serviceNameNB": "JCjR0Lc3J0sm2PcItECdbHXtpCLQCfq3B", + "serviceNameEW": "N8KBAdqT8L_sWOxeFZs3XYn6oykTTFHLiPKOS7kdYbw", + "componentName": "9aCfCEDe2Dv0Peg", + "artefactId": "9c9143f0-f44f-49df-939e-1e8b891ba8f5", + }, + { + "serviceNameNB": "RIfXlfU9cDeLnrOBYzz9LJGdAjwPRp_3Mjp0Wq_RDlQiAPyXm", + "serviceNameEW": "31y8sCwvvyNCXfwtLhwJw6hoblG7ZcFzEjyFdAnzq7M8cxiOtDik0", + "componentName": "3kTa4zKEX", + "artefactId": "9c9143f0-f44f-49df-939e-1e8b891ba8f5", + }, + ], + }, + }, + "aeros": { + # PLACEHOLDER + }, + "kubernetes": { + # PLACEHOLDER + }, +} diff --git a/tests/edgecloud/test_e2e.py b/tests/edgecloud/test_e2e_camara.py similarity index 98% rename from tests/edgecloud/test_e2e.py rename to tests/edgecloud/test_e2e_camara.py index 382a680611bdd861c6540df19e49a24a646f53b8..ec333483b3553b01fdc64e7229edb88bd006cc5f 100644 --- a/tests/edgecloud/test_e2e.py +++ b/tests/edgecloud/test_e2e_camara.py @@ -5,6 +5,7 @@ # Contributors: # - Adrián Pino Martínez (adrian.pino@i2cat.net) # - Sergio Giménez (sergio.gimenez@i2cat.net) +# - César Cajas (cesar.cajas@i2cat.net) ## """ EdgeCloud adapters Integration Tests @@ -31,9 +32,9 @@ from sunrise6g_opensdk.edgecloud.adapters.errors import EdgeCloudPlatformError from sunrise6g_opensdk.edgecloud.adapters.i2edge.client import ( EdgeApplicationManager as I2EdgeClient, ) -from sunrise6g_opensdk.edgecloud.core import schemas as camara_schemas +from sunrise6g_opensdk.edgecloud.core import camara_schemas from tests.edgecloud.test_cases import test_cases -from tests.edgecloud.test_config import CONFIG +from tests.edgecloud.test_config_camara import CONFIG @pytest.fixture(scope="module", name="edgecloud_client") diff --git a/tests/edgecloud/test_e2e_gsma.py b/tests/edgecloud/test_e2e_gsma.py new file mode 100644 index 0000000000000000000000000000000000000000..089d0dc46ac04721ef320bbc39ae66c070ecabcd --- /dev/null +++ b/tests/edgecloud/test_e2e_gsma.py @@ -0,0 +1,368 @@ +# -*- coding: utf-8 -*- +## +# This file is part of the Open SDK +# +# Contributors: +# - Adrián Pino Martínez (adrian.pino@i2cat.net) +# - Sergio Giménez (sergio.gimenez@i2cat.net) +# - César Cajas (cesar.cajas@i2cat.net) +## +""" +EdgeCloud adapters Integration Tests + +Validates the complete application lifecycle: +1. Infrastructure (zone discovery) +2. Artefact management (create/delete) +3. Application lifecycle (onboard/deploy/undeploy/delete app onboarded) + +Key features: +- Tests all client implementations (parametrized via test_cases) +- Tests configuration available in test_config.py +- Ensures proper resource cleanup +- Uses shared test constants and CAMARA-compliant manifests +- Includes artefact unit tests where needed +""" +import time + +import pytest +from requests import Response + +from sunrise6g_opensdk.common.sdk import Sdk as sdkclient +from sunrise6g_opensdk.edgecloud.adapters.errors import EdgeCloudPlatformError +from sunrise6g_opensdk.edgecloud.adapters.i2edge.client import ( + EdgeApplicationManager as I2EdgeClient, +) +from sunrise6g_opensdk.edgecloud.core import gsma_schemas +from tests.edgecloud.test_cases import test_cases +from tests.edgecloud.test_config_gsma import CONFIG + + +@pytest.fixture(scope="module", name="edgecloud_client") +def instantiate_edgecloud_client(request): + """Fixture to create and share an edgecloud client across tests""" + adapter_specs = request.param + client_name = adapter_specs["edgecloud"]["client_name"] + adapters = sdkclient.create_adapters_from(adapter_specs) + client = adapters.get("edgecloud") + client.client_name = client_name + return client + + +def id_func(val): + return val["edgecloud"]["client_name"] + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_config_gsma_compliance(edgecloud_client): + """Validate that all test configurations are GSMA-compliant""" + config = CONFIG[edgecloud_client.client_name] + + try: + # Validate APP_ONBOARD_MANIFEST_GSMA is GSMA-compliant + if "APP_ONBOARD_MANIFEST_GSMA" in config: + app_manifest = config["APP_ONBOARD_MANIFEST_GSMA"] + gsma_schemas.AppOnboardManifestGSMA(**app_manifest) + + # Validate APP_DEPLOY_PAYLOAD_GSMA is GSMA-compliant + if "APP_DEPLOY_PAYLOAD_GSMA" in config: + deploy_payload = config["APP_DEPLOY_PAYLOAD_GSMA"] + gsma_schemas.AppDeployPayloadGSMA(**deploy_payload) + + # Validate PATCH_ONBOARDED_APP_GSMA is GSMA-compliant + if "PATCH_ONBOARDED_APP_GSMA" in config: + patch_payload = config["PATCH_ONBOARDED_APP_GSMA"] + gsma_schemas.PatchOnboardedAppGSMA(**patch_payload) + + except Exception as e: + pytest.fail(f"Configuration is not GSMA-compliant for {edgecloud_client.client_name}: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_get_edge_cloud_zones_list_gsma(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + try: + response = edgecloud_client.get_edge_cloud_zones_list_gsma() + assert isinstance(response, Response) + assert response.status_code == 200 + zones = response.json() + assert isinstance(zones, list) + + # GSMA schema validation for each zone + validated_zones = [] + for zone in zones: + validated_zone = gsma_schemas.ZoneDetails(**zone) + validated_zones.append(validated_zone) + + # Logical validation: verify our expected zone is in the list + expected_zone_id = config["ZONE_ID"] + found_expected_zone = any(str(zone.zoneId) == expected_zone_id for zone in validated_zones) + assert found_expected_zone, f"Expected zone {expected_zone_id} not found in returned zones" + + except EdgeCloudPlatformError as e: + pytest.fail(f"Failed to retrieve zones: {e}") + except Exception as e: + pytest.fail(f"Unexpected error during zone validation: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_get_edge_cloud_zones_gsma(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + try: + response = edgecloud_client.get_edge_cloud_zones_gsma() + assert isinstance(response, Response) + assert response.status_code == 200 + zones = response.json() + assert isinstance(zones, list) + + # GSMA schema validation for each zone + validated_zones = [] + for zone in zones: + validated_zone = gsma_schemas.ZoneRegisteredData(**zone) + validated_zones.append(validated_zone) + + # Logical validation: verify our expected zone is in the list + expected_zone_id = config["ZONE_ID"] + found_expected_zone = any(str(zone.zoneId) == expected_zone_id for zone in validated_zones) + assert found_expected_zone, f"Expected zone {expected_zone_id} not found in returned zones" + + except EdgeCloudPlatformError as e: + pytest.fail(f"Failed to retrieve zones details: {e}") + except Exception as e: + pytest.fail(f"Unexpected error during zone validation: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_get_edge_cloud_zone_details_gsma(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + expected_zone_id = config["ZONE_ID"] + try: + response = edgecloud_client.get_edge_cloud_zone_details_gsma(expected_zone_id) + assert isinstance(response, Response) + assert response.status_code == 200 + zone = response.json() + assert isinstance(zone, dict) + + # GSMA schema validation for zone + validated_zone = gsma_schemas.ZoneRegisteredData(**zone) + + # Logical validation: verify our expected zone is in the dict + assert ( + str(validated_zone.zoneId) == expected_zone_id + ), f"Expected zoneId {expected_zone_id}, got {validated_zone.zoneId}" + + except EdgeCloudPlatformError as e: + pytest.fail(f"Failed to retrieve zones details: {e}") + except Exception as e: + pytest.fail(f"Unexpected error during zone validation: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_artefact_methods_gsma(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + if isinstance(edgecloud_client, I2EdgeClient): + try: + response = edgecloud_client.create_artefact( + artefact_id=config["ARTEFACT_ID"], + artefact_name=config["ARTEFACT_NAME"], + repo_name=config["REPO_NAME"], + repo_type=config["REPO_TYPE"], + repo_url=config["REPO_URL"], + password=None, + token=None, + user_name=None, + ) + assert response.status_code == 201 + except EdgeCloudPlatformError as e: + pytest.fail(f"Artefact creation failed: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_get_artefact_gsma(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + expected_artefact_id = config["ARTEFACT_ID"] + try: + response = edgecloud_client.get_artefact_gsma(expected_artefact_id) + assert isinstance(response, Response) + assert response.status_code == 200 + artefact = response.json() + assert isinstance(artefact, dict) + + # GSMA schema validation for artefact + validated_artefact = gsma_schemas.Artefact(**artefact) + + # Logical validation: verify our expected artefact_id is in the dict + assert ( + str(validated_artefact.artefactId) == expected_artefact_id + ), f"Expected artefactId {expected_artefact_id}, got {validated_artefact.artefactId}" + + except EdgeCloudPlatformError as e: + pytest.fail(f"Failed to retrieve artefact: {e}") + except Exception as e: + pytest.fail(f"Unexpected error during artefact validation: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_onboard_app_gsma(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + try: + response = edgecloud_client.onboard_app_gsma(config["APP_ONBOARD_MANIFEST_GSMA"]) + assert isinstance(response, Response) + assert response.status_code == 200 + + except EdgeCloudPlatformError as e: + pytest.fail(f"App onboarding failed: {e}") + except Exception as e: + pytest.fail(f"Unexpected error during app onboarding: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_get_onboarded_app_gsma(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + app_id = config["APP_ONBOARD_MANIFEST_GSMA"]["appId"] + try: + response = edgecloud_client.get_onboarded_app_gsma(app_id) + assert isinstance(response, Response) + assert response.status_code == 200 + + onboarded_app = response.json() + assert isinstance(onboarded_app, dict) + + # GSMA schema validation for onboarded_app + validated_schema = gsma_schemas.ApplicationModel(**onboarded_app) + + # Logical validation: verify our expected app_id is in the dict + assert ( + str(validated_schema.appId) == app_id + ), f"Expected appId {app_id}, got {validated_schema.appId}" + except EdgeCloudPlatformError as e: + pytest.fail(f"Failed to retrieve app: {e}") + except Exception as e: + pytest.fail(f"Unexpected error validating app: {e}") + + +@pytest.fixture(scope="module") +def app_instance_id_gsma(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + try: + # Use standardized GSMA structure for all adapters + deploy_payload = config["APP_DEPLOY_PAYLOAD_GSMA"] + + response = edgecloud_client.deploy_app_gsma(deploy_payload) + + assert isinstance(response, Response) + assert ( + response.status_code == 202 + ), f"Expected 202, got {response.status_code}: {response.text}" + + response_data = response.json() + instance_info = gsma_schemas.AppInstance(**response_data) + + # Extract appInstIdentifier from the validated object + app_instance_id_gsma = instance_info.appInstIdentifier + + assert app_instance_id_gsma is not None + yield app_instance_id_gsma + finally: + pass + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_deploy_app_gsma(app_instance_id_gsma): + assert app_instance_id_gsma is not None + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_timer_wait_10_seconds(edgecloud_client): + time.sleep(10) + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_get_all_deployed_apps_gsma(edgecloud_client): + """Test retrieving all deployed application instances""" + try: + response = edgecloud_client.get_all_deployed_apps_gsma() + assert isinstance(response, Response) + assert response.status_code == 200 + + instances_data = response.json() + assert isinstance(instances_data, list) + + validated_instances = [] + for instance_data in instances_data: + validated_instance = gsma_schemas.ZoneIdentifier(**instance_data) + validated_instances.append(validated_instance) + + except EdgeCloudPlatformError as e: + pytest.fail(f"Failed to get all deployed apps: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_get_deployed_app_gsma(edgecloud_client, app_instance_id_gsma): + """Test retrieving a specific deployed application instance""" + config = CONFIG[edgecloud_client.client_name] + app_id = config["APP_DEPLOY_PAYLOAD_GSMA"]["appId"] + zone_id = config["APP_DEPLOY_PAYLOAD_GSMA"]["zoneInfo"]["zoneId"] + try: + response = edgecloud_client.get_deployed_app_gsma(app_id, app_instance_id_gsma, zone_id) + assert isinstance(response, Response) + assert response.status_code == 200 + + instance_data = response.json() + assert isinstance(instance_data, dict) + assert "appInstanceState" in instance_data + + gsma_schemas.AppInstanceStatus(**instance_data) + + except EdgeCloudPlatformError as e: + pytest.fail(f"Failed to get deployed app: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_undeploy_app_gsma(edgecloud_client, app_instance_id_gsma): + config = CONFIG[edgecloud_client.client_name] + app_id = config["APP_DEPLOY_PAYLOAD_GSMA"]["appId"] + zone_id = config["APP_DEPLOY_PAYLOAD_GSMA"]["zoneInfo"]["zoneId"] + try: + response = edgecloud_client.undeploy_app_gsma(app_id, app_instance_id_gsma, zone_id) + assert isinstance(response, Response) + assert response.status_code == 200 + except EdgeCloudPlatformError as e: + pytest.fail(f"App undeployment failed: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_patch_onboarded_app_gsma(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + app_id = config["APP_ONBOARD_MANIFEST_GSMA"]["appId"] + try: + payload = config["PATCH_ONBOARDED_APP_GSMA"] + response = edgecloud_client.patch_onboarded_app_gsma(app_id, payload) + assert isinstance(response, Response) + assert response.status_code == 200 + except EdgeCloudPlatformError as e: + pytest.fail(f"Failed to patch onboarded app: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_delete_onboarded_app_gsma(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + try: + app_id = config["APP_ONBOARD_MANIFEST_GSMA"]["appId"] + response = edgecloud_client.delete_onboarded_app_gsma(app_id) + assert isinstance(response, Response) + assert response.status_code == 200 + except EdgeCloudPlatformError as e: + pytest.fail(f"App onboarding deletion failed: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_delete_artefact_gsma(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + + if isinstance(edgecloud_client, I2EdgeClient): + try: + response = edgecloud_client.delete_artefact_gsma(config["ARTEFACT_ID"]) + assert isinstance(response, Response) + assert response.status_code == 200 + except EdgeCloudPlatformError as e: + pytest.fail(f"Artefact deletion failed: {e}")