diff --git a/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py b/src/sunrise6g_opensdk/edgecloud/core/gsma_schemas.py index c2ac41afb4cef4d1736e1e2e2c04fa3eabac1b0e..01082be91552ed469482a9f0d7f83d8a69916e77 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 156ff46ca8a08277031c53a4018eb59d3c3dcd39..341b899ce6799b2d9ef08c5a23a9bdbb81347381 100644 --- a/tests/edgecloud/test_config.py +++ b/tests/edgecloud/test_config.py @@ -65,6 +65,81 @@ 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 diff --git a/tests/edgecloud/test_e2e.py b/tests/edgecloud/test_e2e.py index 382a680611bdd861c6540df19e49a24a646f53b8..9af8922d3b40a4d1aee912e9bf566129416e3d8f 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}")