Loading src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py +4 −4 Original line number Diff line number Diff line Loading @@ -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 Loading Loading @@ -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 Loading tests/edgecloud/test_config.py +116 −48 Original line number Diff line number Diff line Loading @@ -41,7 +41,7 @@ CONFIG = { "componentName": "my-component", "networkInterfaces": [ { "interfaceId": "c3e1e6d4-1a5c-4a3b-9e4a-5f3d7b6c8e01", "interfaceId": "eth0", "protocol": "TCP", "port": 8080, "visibilityType": "VISIBILITY_EXTERNAL", Loading @@ -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": "<UUID>", "APP_ID": "<UUID>", # 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": "<UUID>", "appId": "aeros-app-2", "name": "aeros-SDK-app", "version": "1.0.0", "appProvider": "aeros", Loading Loading @@ -102,7 +179,7 @@ CONFIG = { "componentName": "aeros-component", "networkInterfaces": [ { "interfaceId": "<UUID>", "interfaceId": "eth0", "protocol": "TCP", "port": 9090, "visibilityType": "VISIBILITY_INTERNAL", Loading @@ -111,13 +188,14 @@ CONFIG = { } ], }, "APP_ID": "aeros-app-2", # CAMARA deploy_app payload "APP_DEPLOY_PAYLOAD": { "appId": "<UUID>", "appId": "aeros-app-2", "appZones": [ { "EdgeCloudZone": { "edgeCloudZoneId": "<UUID>", "edgeCloudZoneId": "urn:ngsi-ld:Domain:NCSRD", "edgeCloudZoneName": "aeros-zone-1", "edgeCloudZoneStatus": "active", "edgeCloudProvider": "NCSRD", Loading @@ -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", Loading @@ -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", } } ], }, }, } tests/edgecloud/test_e2e.py +298 −0 Original line number Diff line number Diff line Loading @@ -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 Loading @@ -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 Loading Loading @@ -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}") Loading
src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py +4 −4 Original line number Diff line number Diff line Loading @@ -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 Loading Loading @@ -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 Loading
tests/edgecloud/test_config.py +116 −48 Original line number Diff line number Diff line Loading @@ -41,7 +41,7 @@ CONFIG = { "componentName": "my-component", "networkInterfaces": [ { "interfaceId": "c3e1e6d4-1a5c-4a3b-9e4a-5f3d7b6c8e01", "interfaceId": "eth0", "protocol": "TCP", "port": 8080, "visibilityType": "VISIBILITY_EXTERNAL", Loading @@ -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": "<UUID>", "APP_ID": "<UUID>", # 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": "<UUID>", "appId": "aeros-app-2", "name": "aeros-SDK-app", "version": "1.0.0", "appProvider": "aeros", Loading Loading @@ -102,7 +179,7 @@ CONFIG = { "componentName": "aeros-component", "networkInterfaces": [ { "interfaceId": "<UUID>", "interfaceId": "eth0", "protocol": "TCP", "port": 9090, "visibilityType": "VISIBILITY_INTERNAL", Loading @@ -111,13 +188,14 @@ CONFIG = { } ], }, "APP_ID": "aeros-app-2", # CAMARA deploy_app payload "APP_DEPLOY_PAYLOAD": { "appId": "<UUID>", "appId": "aeros-app-2", "appZones": [ { "EdgeCloudZone": { "edgeCloudZoneId": "<UUID>", "edgeCloudZoneId": "urn:ngsi-ld:Domain:NCSRD", "edgeCloudZoneName": "aeros-zone-1", "edgeCloudZoneStatus": "active", "edgeCloudProvider": "NCSRD", Loading @@ -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", Loading @@ -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", } } ], }, }, }
tests/edgecloud/test_e2e.py +298 −0 Original line number Diff line number Diff line Loading @@ -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 Loading @@ -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 Loading Loading @@ -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}")
