From 5f33e7876657a9d2f5fe55255af82bdf92e51551 Mon Sep 17 00:00:00 2001 From: cesarcajas Date: Wed, 13 Aug 2025 09:50:21 +0200 Subject: [PATCH 1/3] feature/add-e2e-tests-gsma-edgecloud: add tests e2e edgecloud gsma methods --- .../edgecloud/core/gsma_schemas.py | 8 +- tests/edgecloud/test_config.py | 164 +++++++--- tests/edgecloud/test_e2e.py | 298 ++++++++++++++++++ 3 files changed, 418 insertions(+), 52 deletions(-) diff --git a/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py b/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py index c2ac41a..01082be 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 diff --git a/tests/edgecloud/test_config.py b/tests/edgecloud/test_config.py index 156ff46..67cd84d 100644 --- a/tests/edgecloud/test_config.py +++ b/tests/edgecloud/test_config.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", @@ -65,14 +65,91 @@ CONFIG = { } ], }, + # GSMA + "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": "67f3a0b0e3184a85952e174d", + "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": { - # Basic identifiers - "ZONE_ID": "", - "APP_ID": "", - # CAMARA onboard_app payload + "ZONE_ID": "urn:ngsi-ld:Domain:NCSRD", + "ARTEFACT_ID": "aeros-app-2", + "ARTEFACT_NAME": "aeroschart", + "REPO_NAME": "github-aeros", + "REPO_TYPE": "PUBLICREPO", + "REPO_URL": "https://aeros.github.io/helm/", "APP_ONBOARD_MANIFEST": { - "appId": "", + "appId": "aeros-app-2", "name": "aeros-SDK-app", "version": "1.0.0", "appProvider": "aeros", @@ -102,7 +179,7 @@ CONFIG = { "componentName": "aeros-component", "networkInterfaces": [ { - "interfaceId": "", + "interfaceId": "eth0", "protocol": "TCP", "port": 9090, "visibilityType": "VISIBILITY_INTERNAL", @@ -111,13 +188,14 @@ CONFIG = { } ], }, + "APP_ID": "aeros-app-2", # CAMARA deploy_app payload "APP_DEPLOY_PAYLOAD": { - "appId": "", + "appId": "aeros-app-2", "appZones": [ { "EdgeCloudZone": { - "edgeCloudZoneId": "", + "edgeCloudZoneId": "urn:ngsi-ld:Domain:NCSRD", "edgeCloudZoneName": "aeros-zone-1", "edgeCloudZoneStatus": "active", "edgeCloudProvider": "NCSRD", @@ -128,34 +206,39 @@ CONFIG = { }, }, "kubernetes": { - # Basic identifiers - "ZONE_ID": "999b7746-d2e2-4bb4-96e6-f1e895adef0c", + "K8S_ONBOARDED_APP_NAME": "nginx", + "K8S_APP_ID": "3fa85f64-5717-4562-b3fc-2c963f66afa6", "APP_ID": "3fa85f64-5717-4562-b3fc-2c963f66afa6", - "ZONE_ID": "999b7746-d2e2-4bb4-96e6-f1e895adef0c", - # CAMARA onboard_app payload + "ZONE_ID": "b2a1b33d-f382-47de-b555-2d32155eb74c", + # CAMARA deploy_app payload + "APP_DEPLOY_PAYLOAD": { + "appId": "3fa85f64-5717-4562-b3fc-2c963f66afa6", + "appZones": [ + { + "EdgeCloudZone": { + "edgeCloudZoneId": "b2a1b33d-f382-47de-b555-2d32155eb74c", + "edgeCloudZoneName": "k8s-zone-1", + "edgeCloudZoneStatus": "active", + "edgeCloudProvider": "kubernetes", + "edgeCloudRegion": "Local", + } + } + ], + }, + # Legacy K8S_DEPLOY_PAYLOAD for backward compatibility (if needed) + "K8S_DEPLOY_PAYLOAD": { + "appId": "3fa85f64-5717-4562-b3fc-2c963f66afa6", + "name": "nginx-test", + "edgeCloudZoneId": "zorro-solutions", + "kubernetesClusterRef": "", + }, "APP_ONBOARD_MANIFEST": { "appId": "3fa85f64-5717-4562-b3fc-2c963f66afa6", - "name": "nginx_test", + "name": "nginx", "version": "1", "packageType": "QCOW2", - "appProvider": "jJsxUU403g5PUhoRtYfOBaaIRZGVzXDZgMBqzMYq5xLcdZoWGYIYqmVy0", + "appProvider": "Nginx Inc.", "appRepo": {"imagePath": "nginx", "type": "PRIVATEREPO"}, - "requiredResources": { - "infraKind": "kubernetes", - "applicationResources": { - "cpuPool": { - "numCPU": 2, - "memory": 2048, - "topology": { - "minNumberOfNodes": 2, - "minNodeCpu": 1, - "minNodeMemory": 1024, - }, - } - }, - "isStandalone": False, - "version": "1.29", - }, "componentSpec": [ { "componentName": "nginx", @@ -163,33 +246,18 @@ CONFIG = { { "protocol": "TCP", "port": 80, - "interfaceId": "8f4b02a9-73ba-4fde-9d93-3b1a6e7c5d9f", + "interfaceId": "Uj6qThvzkegxa3L4b88", "visibilityType": "VISIBILITY_EXTERNAL", }, { "protocol": "TCP", "port": 443, - "interfaceId": "e5c9a2b1-3d7f-4b8e-a6c4-1f9d8b7a6c0e", + "interfaceId": "Uj6qThvzkegxa3L4b88", "visibilityType": "VISIBILITY_EXTERNAL", }, ], } ], }, - # CAMARA deploy_app payload - "APP_DEPLOY_PAYLOAD": { - "appId": "3fa85f64-5717-4562-b3fc-2c963f66afa6", - "appZones": [ - { - "EdgeCloudZone": { - "edgeCloudZoneId": "999b7746-d2e2-4bb4-96e6-f1e895adef0c", - "edgeCloudZoneName": "zorro-solutions", - "edgeCloudZoneStatus": "active", - "edgeCloudProvider": "ICOM", - "edgeCloudRegion": "Europe-Southeast", - } - } - ], - }, }, } diff --git a/tests/edgecloud/test_e2e.py b/tests/edgecloud/test_e2e.py index 382a680..9af8922 100644 --- a/tests/edgecloud/test_e2e.py +++ b/tests/edgecloud/test_e2e.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,6 +32,7 @@ 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 sunrise6g_opensdk.edgecloud.core import schemas as camara_schemas from tests.edgecloud.test_cases import test_cases from tests.edgecloud.test_config import CONFIG @@ -338,3 +340,299 @@ def test_delete_artefact(edgecloud_client): edgecloud_client.delete_artefact(artefact_id=config["ARTEFACT_ID"]) except EdgeCloudPlatformError as e: pytest.fail(f"Artefact deletion failed: {e}") + + +# ==================================================================== +# GSMA EDGE COMPUTING API (EWBI OPG) - FEDERATION +# ==================================================================== + + +@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}") -- GitLab From 4e11f16eb67a0fcd89334192d755f6d9d5bee161 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adri=C3=A1n=20Pino?= Date: Fri, 15 Aug 2025 20:08:51 +0200 Subject: [PATCH 2/3] Separate CAMARA & GSMA tests. Update schemas to camara_schemas. --- .../edgecloud/adapters/i2edge/client.py | 3 +- .../edgecloud/adapters/kubernetes/client.py | 2 +- .../core/{schemas.py => camara_schemas.py} | 0 .../{test_config.py => test_config_camara.py} | 160 +++----- tests/edgecloud/test_config_gsma.py | 84 +++++ tests/edgecloud/test_e2e_camara.py | 341 ++++++++++++++++++ .../{test_e2e.py => test_e2e_gsma.py} | 296 +-------------- 7 files changed, 476 insertions(+), 410 deletions(-) rename src/sunrise6g_opensdk/edgecloud/core/{schemas.py => camara_schemas.py} (100%) rename tests/edgecloud/{test_config.py => test_config_camara.py} (55%) create mode 100644 tests/edgecloud/test_config_gsma.py create mode 100644 tests/edgecloud/test_e2e_camara.py rename tests/edgecloud/{test_e2e.py => test_e2e_gsma.py} (54%) diff --git a/src/sunrise6g_opensdk/edgecloud/adapters/i2edge/client.py b/src/sunrise6g_opensdk/edgecloud/adapters/i2edge/client.py index 336fc34..70fa247 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 a02d85d..c315851 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/tests/edgecloud/test_config.py b/tests/edgecloud/test_config_camara.py similarity index 55% rename from tests/edgecloud/test_config.py rename to tests/edgecloud/test_config_camara.py index 67cd84d..6ce39c2 100644 --- a/tests/edgecloud/test_config.py +++ b/tests/edgecloud/test_config_camara.py @@ -65,91 +65,14 @@ CONFIG = { } ], }, - # GSMA - "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": "67f3a0b0e3184a85952e174d", - "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": { - "ZONE_ID": "urn:ngsi-ld:Domain:NCSRD", - "ARTEFACT_ID": "aeros-app-2", - "ARTEFACT_NAME": "aeroschart", - "REPO_NAME": "github-aeros", - "REPO_TYPE": "PUBLICREPO", - "REPO_URL": "https://aeros.github.io/helm/", + # Basic identifiers + "ZONE_ID": "", + "APP_ID": "", + # CAMARA onboard_app payload "APP_ONBOARD_MANIFEST": { - "appId": "aeros-app-2", + "appId": "", "name": "aeros-SDK-app", "version": "1.0.0", "appProvider": "aeros", @@ -188,14 +111,13 @@ CONFIG = { } ], }, - "APP_ID": "aeros-app-2", # CAMARA deploy_app payload "APP_DEPLOY_PAYLOAD": { - "appId": "aeros-app-2", + "appId": "", "appZones": [ { "EdgeCloudZone": { - "edgeCloudZoneId": "urn:ngsi-ld:Domain:NCSRD", + "edgeCloudZoneId": "", "edgeCloudZoneName": "aeros-zone-1", "edgeCloudZoneStatus": "active", "edgeCloudProvider": "NCSRD", @@ -206,39 +128,34 @@ CONFIG = { }, }, "kubernetes": { - "K8S_ONBOARDED_APP_NAME": "nginx", - "K8S_APP_ID": "3fa85f64-5717-4562-b3fc-2c963f66afa6", + # Basic identifiers + "ZONE_ID": "999b7746-d2e2-4bb4-96e6-f1e895adef0c", "APP_ID": "3fa85f64-5717-4562-b3fc-2c963f66afa6", - "ZONE_ID": "b2a1b33d-f382-47de-b555-2d32155eb74c", - # CAMARA deploy_app payload - "APP_DEPLOY_PAYLOAD": { - "appId": "3fa85f64-5717-4562-b3fc-2c963f66afa6", - "appZones": [ - { - "EdgeCloudZone": { - "edgeCloudZoneId": "b2a1b33d-f382-47de-b555-2d32155eb74c", - "edgeCloudZoneName": "k8s-zone-1", - "edgeCloudZoneStatus": "active", - "edgeCloudProvider": "kubernetes", - "edgeCloudRegion": "Local", - } - } - ], - }, - # Legacy K8S_DEPLOY_PAYLOAD for backward compatibility (if needed) - "K8S_DEPLOY_PAYLOAD": { - "appId": "3fa85f64-5717-4562-b3fc-2c963f66afa6", - "name": "nginx-test", - "edgeCloudZoneId": "zorro-solutions", - "kubernetesClusterRef": "", - }, + "ZONE_ID": "999b7746-d2e2-4bb4-96e6-f1e895adef0c", + # CAMARA onboard_app payload "APP_ONBOARD_MANIFEST": { "appId": "3fa85f64-5717-4562-b3fc-2c963f66afa6", - "name": "nginx", + "name": "nginx_test", "version": "1", "packageType": "QCOW2", - "appProvider": "Nginx Inc.", + "appProvider": "jJsxUU403g5PUhoRtYfOBaaIRZGVzXDZgMBqzMYq5xLcdZoWGYIYqmVy0", "appRepo": {"imagePath": "nginx", "type": "PRIVATEREPO"}, + "requiredResources": { + "infraKind": "kubernetes", + "applicationResources": { + "cpuPool": { + "numCPU": 2, + "memory": 2048, + "topology": { + "minNumberOfNodes": 2, + "minNodeCpu": 1, + "minNodeMemory": 1024, + }, + } + }, + "isStandalone": False, + "version": "1.29", + }, "componentSpec": [ { "componentName": "nginx", @@ -246,18 +163,33 @@ CONFIG = { { "protocol": "TCP", "port": 80, - "interfaceId": "Uj6qThvzkegxa3L4b88", + "interfaceId": "http_interface", "visibilityType": "VISIBILITY_EXTERNAL", }, { "protocol": "TCP", "port": 443, - "interfaceId": "Uj6qThvzkegxa3L4b88", + "interfaceId": "https_interface", "visibilityType": "VISIBILITY_EXTERNAL", }, ], } ], }, + # CAMARA deploy_app payload + "APP_DEPLOY_PAYLOAD": { + "appId": "3fa85f64-5717-4562-b3fc-2c963f66afa6", + "appZones": [ + { + "EdgeCloudZone": { + "edgeCloudZoneId": "999b7746-d2e2-4bb4-96e6-f1e895adef0c", + "edgeCloudZoneName": "zorro-solutions", + "edgeCloudZoneStatus": "active", + "edgeCloudProvider": "ICOM", + "edgeCloudRegion": "Europe-Southeast", + } + } + ], + }, }, } diff --git a/tests/edgecloud/test_config_gsma.py b/tests/edgecloud/test_config_gsma.py new file mode 100644 index 0000000..7d58d1c --- /dev/null +++ b/tests/edgecloud/test_config_gsma.py @@ -0,0 +1,84 @@ +CONFIG = { + "i2edge": { + "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": "67f3a0b0e3184a85952e174d", + "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_camara.py b/tests/edgecloud/test_e2e_camara.py new file mode 100644 index 0000000..ec33348 --- /dev/null +++ b/tests/edgecloud/test_e2e_camara.py @@ -0,0 +1,341 @@ +# -*- 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 camara_schemas +from tests.edgecloud.test_cases import test_cases +from tests.edgecloud.test_config_camara 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_camara_compliance(edgecloud_client): + """Validate that all test configurations are CAMARA-compliant""" + config = CONFIG[edgecloud_client.client_name] + + try: + # Validate APP_ONBOARD_MANIFEST is CAMARA-compliant + if "APP_ONBOARD_MANIFEST" in config: + app_manifest = config["APP_ONBOARD_MANIFEST"] + camara_schemas.AppManifest(**app_manifest) + + # Validate APP_DEPLOY_PAYLOAD is CAMARA-compliant + if "APP_DEPLOY_PAYLOAD" in config: + deploy_payload = config["APP_DEPLOY_PAYLOAD"] + + # Validate appId + assert "appId" in deploy_payload + camara_schemas.AppId(root=deploy_payload["appId"]) + + # Validate appZones structure + assert "appZones" in deploy_payload + assert isinstance(deploy_payload["appZones"], list) + assert len(deploy_payload["appZones"]) > 0 + + for zone_data in deploy_payload["appZones"]: + assert "EdgeCloudZone" in zone_data + edge_cloud_zone = zone_data["EdgeCloudZone"] + camara_schemas.EdgeCloudZone( + **edge_cloud_zone + ) # Validate against CAMARA EdgeCloudZone schema + + # Validate APP_ID is consistent + if "APP_ID" in config: + app_id = config["APP_ID"] + camara_schemas.AppId(root=app_id) + + # Check consistency between APP_ID and manifest/payload + if "APP_ONBOARD_MANIFEST" in config: + assert config["APP_ONBOARD_MANIFEST"]["appId"] == app_id + if "APP_DEPLOY_PAYLOAD" in config: + assert config["APP_DEPLOY_PAYLOAD"]["appId"] == app_id + + except Exception as e: + pytest.fail( + f"Configuration is not CAMARA-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(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + try: + response = edgecloud_client.get_edge_cloud_zones() + assert isinstance(response, Response) + assert response.status_code == 200 + zones = response.json() + assert isinstance(zones, list) + + # CAMARA schema validation for each zone + validated_zones = [] + for zone in zones: + validated_zone = camara_schemas.EdgeCloudZone(**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.edgeCloudZoneId.root) == 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_create_artefact(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + if isinstance(edgecloud_client, I2EdgeClient): + try: + 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, + ) + 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_onboard_app(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + try: + response = edgecloud_client.onboard_app(config["APP_ONBOARD_MANIFEST"]) + assert isinstance(response, Response) + assert response.status_code == 201 + + payload = response.json() + assert isinstance(payload, dict) + + # Use CAMARA schema validation for submitted app response + submitted_app = camara_schemas.SubmittedApp(**payload) + assert submitted_app.appId.root == config["APP_ID"] + + 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.fixture(scope="module") +def app_instance_id(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + try: + # Use standardized CAMARA structure for all adapters + deploy_payload = config["APP_DEPLOY_PAYLOAD"] + app_id = deploy_payload["appId"] + app_zones = deploy_payload["appZones"] + + # edgecloud_client.deploy_app maps with CAMARA POST /appinstances + response = edgecloud_client.deploy_app(app_id, app_zones) + + assert isinstance(response, Response) + assert ( + response.status_code == 202 + ), f"Expected 202, got {response.status_code}: {response.text}" + + response_data = response.json() + instance_info = camara_schemas.AppInstanceInfo(**response_data) + + # Extract appInstanceId from the validated object + app_instance_id = instance_info.appInstanceId.root + + assert app_instance_id is not None + yield app_instance_id + finally: + pass + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_deploy_app(app_instance_id): + assert app_instance_id is not None + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_timer_wait_30_seconds(edgecloud_client): + time.sleep(30) + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_get_onboarded_app(edgecloud_client): + """Test retrieving a specific onboarded application""" + config = CONFIG[edgecloud_client.client_name] + try: + app_id = config["APP_ID"] + response = edgecloud_client.get_onboarded_app(app_id) + assert isinstance(response, Response) + assert response.status_code == 200 + + app_data = response.json() + assert isinstance(app_data, dict) + assert "appManifest" in app_data + + # Use CAMARA schema validation instead of manual checks + app_manifest = camara_schemas.AppManifest(**app_data["appManifest"]) + assert app_manifest.appId.root == app_id + + except EdgeCloudPlatformError as e: + pytest.fail(f"Failed to get onboarded app: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_get_all_onboarded_apps(edgecloud_client): + """Test retrieving all onboarded applications""" + config = CONFIG[edgecloud_client.client_name] + try: + response = edgecloud_client.get_all_onboarded_apps() + assert isinstance(response, Response) + assert response.status_code == 200 + + apps_data = response.json() + assert isinstance(apps_data, list) + + # CAMARA schema validation for each app manifest + validated_apps = [] + for app_manifest_data in apps_data: + validated_app = camara_schemas.AppManifest(**app_manifest_data) + validated_apps.append(validated_app) + + # Logical validation: verify our onboarded app is in the list + expected_app_id = config["APP_ID"] + found_expected_app = any(str(app.appId.root) == expected_app_id for app in validated_apps) + assert found_expected_app, f"Expected app {expected_app_id} not found in onboarded apps" + + except EdgeCloudPlatformError as e: + pytest.fail(f"Failed to get all onboarded apps: {e}") + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_get_all_deployed_apps(edgecloud_client): + """Test retrieving all deployed application instances""" + try: + response = edgecloud_client.get_all_deployed_apps() + assert isinstance(response, Response) + assert response.status_code == 200 + + instances_data = response.json() + assert isinstance(instances_data, dict) + assert "appInstances" in instances_data + assert isinstance(instances_data["appInstances"], list) + + # CAMARA schema validation for each app instance + validated_instances = [] + for instance_data in instances_data["appInstances"]: + validated_instance = camara_schemas.AppInstanceInfo(**instance_data) + validated_instances.append(validated_instance) + + # TODO: validate that the newly created app instance is in the list + + 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(edgecloud_client, app_instance_id): + """Test retrieving a specific deployed application instance""" + try: + response = edgecloud_client.get_deployed_app(app_instance_id) + assert isinstance(response, Response) + assert response.status_code == 200 + + instance_data = response.json() + assert isinstance(instance_data, dict) + assert "appInstance" in instance_data + + # Use CAMARA schema validation for the app instance + app_instance = camara_schemas.AppInstanceInfo(**instance_data["appInstance"]) + assert app_instance.appInstanceId.root == app_instance_id + + # TODO: validate that we can get the newly created app + + 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(edgecloud_client, app_instance_id): + try: + response = edgecloud_client.undeploy_app(app_instance_id) + assert isinstance(response, Response) + assert response.status_code == 204 + assert response.text == "" + 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_timer_wait_5_seconds(edgecloud_client): +# time.sleep(5) + + +@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) +def test_delete_onboarded_app(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + try: + app_id = config["APP_ID"] + edgecloud_client.delete_onboarded_app(app_id=app_id) + 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(edgecloud_client): + config = CONFIG[edgecloud_client.client_name] + + if isinstance(edgecloud_client, I2EdgeClient): + try: + edgecloud_client.delete_artefact(artefact_id=config["ARTEFACT_ID"]) + except EdgeCloudPlatformError as e: + pytest.fail(f"Artefact deletion failed: {e}") diff --git a/tests/edgecloud/test_e2e.py b/tests/edgecloud/test_e2e_gsma.py similarity index 54% rename from tests/edgecloud/test_e2e.py rename to tests/edgecloud/test_e2e_gsma.py index 9af8922..8939d0c 100644 --- a/tests/edgecloud/test_e2e.py +++ b/tests/edgecloud/test_e2e_gsma.py @@ -33,9 +33,8 @@ from sunrise6g_opensdk.edgecloud.adapters.i2edge.client import ( EdgeApplicationManager as I2EdgeClient, ) from sunrise6g_opensdk.edgecloud.core import gsma_schemas -from sunrise6g_opensdk.edgecloud.core import schemas as camara_schemas from tests.edgecloud.test_cases import test_cases -from tests.edgecloud.test_config import CONFIG +from tests.edgecloud.test_config_gsma import CONFIG @pytest.fixture(scope="module", name="edgecloud_client") @@ -53,298 +52,9 @@ def id_func(val): return val["edgecloud"]["client_name"] -@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) -def test_config_camara_compliance(edgecloud_client): - """Validate that all test configurations are CAMARA-compliant""" - config = CONFIG[edgecloud_client.client_name] - - try: - # Validate APP_ONBOARD_MANIFEST is CAMARA-compliant - if "APP_ONBOARD_MANIFEST" in config: - app_manifest = config["APP_ONBOARD_MANIFEST"] - camara_schemas.AppManifest(**app_manifest) - - # Validate APP_DEPLOY_PAYLOAD is CAMARA-compliant - if "APP_DEPLOY_PAYLOAD" in config: - deploy_payload = config["APP_DEPLOY_PAYLOAD"] - - # Validate appId - assert "appId" in deploy_payload - camara_schemas.AppId(root=deploy_payload["appId"]) - - # Validate appZones structure - assert "appZones" in deploy_payload - assert isinstance(deploy_payload["appZones"], list) - assert len(deploy_payload["appZones"]) > 0 - - for zone_data in deploy_payload["appZones"]: - assert "EdgeCloudZone" in zone_data - edge_cloud_zone = zone_data["EdgeCloudZone"] - camara_schemas.EdgeCloudZone( - **edge_cloud_zone - ) # Validate against CAMARA EdgeCloudZone schema - - # Validate APP_ID is consistent - if "APP_ID" in config: - app_id = config["APP_ID"] - camara_schemas.AppId(root=app_id) - - # Check consistency between APP_ID and manifest/payload - if "APP_ONBOARD_MANIFEST" in config: - assert config["APP_ONBOARD_MANIFEST"]["appId"] == app_id - if "APP_DEPLOY_PAYLOAD" in config: - assert config["APP_DEPLOY_PAYLOAD"]["appId"] == app_id - - except Exception as e: - pytest.fail( - f"Configuration is not CAMARA-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(edgecloud_client): - config = CONFIG[edgecloud_client.client_name] - try: - response = edgecloud_client.get_edge_cloud_zones() - assert isinstance(response, Response) - assert response.status_code == 200 - zones = response.json() - assert isinstance(zones, list) - - # CAMARA schema validation for each zone - validated_zones = [] - for zone in zones: - validated_zone = camara_schemas.EdgeCloudZone(**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.edgeCloudZoneId.root) == 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_create_artefact(edgecloud_client): - config = CONFIG[edgecloud_client.client_name] - if isinstance(edgecloud_client, I2EdgeClient): - try: - 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, - ) - 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_onboard_app(edgecloud_client): - config = CONFIG[edgecloud_client.client_name] - try: - response = edgecloud_client.onboard_app(config["APP_ONBOARD_MANIFEST"]) - assert isinstance(response, Response) - assert response.status_code == 201 - - payload = response.json() - assert isinstance(payload, dict) - - # Use CAMARA schema validation for submitted app response - submitted_app = camara_schemas.SubmittedApp(**payload) - assert submitted_app.appId.root == config["APP_ID"] - - 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.fixture(scope="module") -def app_instance_id(edgecloud_client): - config = CONFIG[edgecloud_client.client_name] - try: - # Use standardized CAMARA structure for all adapters - deploy_payload = config["APP_DEPLOY_PAYLOAD"] - app_id = deploy_payload["appId"] - app_zones = deploy_payload["appZones"] - - # edgecloud_client.deploy_app maps with CAMARA POST /appinstances - response = edgecloud_client.deploy_app(app_id, app_zones) - - assert isinstance(response, Response) - assert ( - response.status_code == 202 - ), f"Expected 202, got {response.status_code}: {response.text}" - - response_data = response.json() - instance_info = camara_schemas.AppInstanceInfo(**response_data) - - # Extract appInstanceId from the validated object - app_instance_id = instance_info.appInstanceId.root - - assert app_instance_id is not None - yield app_instance_id - finally: - pass - - -@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) -def test_deploy_app(app_instance_id): - assert app_instance_id is not None - - -@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) -def test_timer_wait_30_seconds(edgecloud_client): - time.sleep(30) - - -@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) -def test_get_onboarded_app(edgecloud_client): - """Test retrieving a specific onboarded application""" - config = CONFIG[edgecloud_client.client_name] - try: - app_id = config["APP_ID"] - response = edgecloud_client.get_onboarded_app(app_id) - assert isinstance(response, Response) - assert response.status_code == 200 - - app_data = response.json() - assert isinstance(app_data, dict) - assert "appManifest" in app_data - - # Use CAMARA schema validation instead of manual checks - app_manifest = camara_schemas.AppManifest(**app_data["appManifest"]) - assert app_manifest.appId.root == app_id - - except EdgeCloudPlatformError as e: - pytest.fail(f"Failed to get onboarded app: {e}") - - -@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) -def test_get_all_onboarded_apps(edgecloud_client): - """Test retrieving all onboarded applications""" - config = CONFIG[edgecloud_client.client_name] - try: - response = edgecloud_client.get_all_onboarded_apps() - assert isinstance(response, Response) - assert response.status_code == 200 - - apps_data = response.json() - assert isinstance(apps_data, list) - - # CAMARA schema validation for each app manifest - validated_apps = [] - for app_manifest_data in apps_data: - validated_app = camara_schemas.AppManifest(**app_manifest_data) - validated_apps.append(validated_app) - - # Logical validation: verify our onboarded app is in the list - expected_app_id = config["APP_ID"] - found_expected_app = any(str(app.appId.root) == expected_app_id for app in validated_apps) - assert found_expected_app, f"Expected app {expected_app_id} not found in onboarded apps" - - except EdgeCloudPlatformError as e: - pytest.fail(f"Failed to get all onboarded apps: {e}") - - -@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) -def test_get_all_deployed_apps(edgecloud_client): - """Test retrieving all deployed application instances""" - try: - response = edgecloud_client.get_all_deployed_apps() - assert isinstance(response, Response) - assert response.status_code == 200 - - instances_data = response.json() - assert isinstance(instances_data, dict) - assert "appInstances" in instances_data - assert isinstance(instances_data["appInstances"], list) - - # CAMARA schema validation for each app instance - validated_instances = [] - for instance_data in instances_data["appInstances"]: - validated_instance = camara_schemas.AppInstanceInfo(**instance_data) - validated_instances.append(validated_instance) - - # TODO: validate that the newly created app instance is in the list - - 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(edgecloud_client, app_instance_id): - """Test retrieving a specific deployed application instance""" - try: - response = edgecloud_client.get_deployed_app(app_instance_id) - assert isinstance(response, Response) - assert response.status_code == 200 - - instance_data = response.json() - assert isinstance(instance_data, dict) - assert "appInstance" in instance_data - - # Use CAMARA schema validation for the app instance - app_instance = camara_schemas.AppInstanceInfo(**instance_data["appInstance"]) - assert app_instance.appInstanceId.root == app_instance_id - - # TODO: validate that we can get the newly created app - - 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(edgecloud_client, app_instance_id): - try: - response = edgecloud_client.undeploy_app(app_instance_id) - assert isinstance(response, Response) - assert response.status_code == 204 - assert response.text == "" - except EdgeCloudPlatformError as e: - pytest.fail(f"App undeployment failed: {e}") - - +# TODO # @pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) -# def test_timer_wait_5_seconds(edgecloud_client): -# time.sleep(5) - - -@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) -def test_delete_onboarded_app(edgecloud_client): - config = CONFIG[edgecloud_client.client_name] - try: - app_id = config["APP_ID"] - edgecloud_client.delete_onboarded_app(app_id=app_id) - 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(edgecloud_client): - config = CONFIG[edgecloud_client.client_name] - - if isinstance(edgecloud_client, I2EdgeClient): - try: - edgecloud_client.delete_artefact(artefact_id=config["ARTEFACT_ID"]) - except EdgeCloudPlatformError as e: - pytest.fail(f"Artefact deletion failed: {e}") - - -# ==================================================================== -# GSMA EDGE COMPUTING API (EWBI OPG) - FEDERATION -# ==================================================================== +# def test_config_gsma_compliance(edgecloud_client): @pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) -- GitLab From 828e4e365b311bee91adfba9d00be7f70a7de840 Mon Sep 17 00:00:00 2001 From: cesarcajas Date: Mon, 8 Sep 2025 14:25:05 +0200 Subject: [PATCH 3/3] feature/add-e2e-tests-gsma-edgecloud: add gsma compliance test, verify all gsma tests --- .../edgecloud/core/gsma_schemas.py | 62 +++++++++++++++++++ tests/edgecloud/test_config_gsma.py | 8 ++- tests/edgecloud/test_e2e_gsma.py | 26 +++++++- 3 files changed, 92 insertions(+), 4 deletions(-) diff --git a/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py b/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py index 01082be..1a1a661 100644 --- a/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py +++ b/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py @@ -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_gsma.py b/tests/edgecloud/test_config_gsma.py index 7d58d1c..94fea67 100644 --- a/tests/edgecloud/test_config_gsma.py +++ b/tests/edgecloud/test_config_gsma.py @@ -1,5 +1,11 @@ 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", @@ -38,7 +44,7 @@ CONFIG = { "appProviderId": "Y89TSlxMPDKlXZz7rN6vU2y", "zoneInfo": { "zoneId": "f0662bfe-1d90-5f59-a759-c755b3b69b93", - "flavourId": "67f3a0b0e3184a85952e174d", + "flavourId": "6881e358535a2eaedcb27214", "resourceConsumption": "RESERVED_RES_AVOID", "resPool": "ySIT0LuZ6ApHs0wlyGZve", }, diff --git a/tests/edgecloud/test_e2e_gsma.py b/tests/edgecloud/test_e2e_gsma.py index 8939d0c..089d0dc 100644 --- a/tests/edgecloud/test_e2e_gsma.py +++ b/tests/edgecloud/test_e2e_gsma.py @@ -52,9 +52,29 @@ def id_func(val): return val["edgecloud"]["client_name"] -# TODO -# @pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True) -# def test_config_gsma_compliance(edgecloud_client): +@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) -- GitLab