Commit 4d3661a0 authored by Cesar Cajas's avatar Cesar Cajas
Browse files

Merge branch 'main' into feature/add-gsma-schemas-to-edgecloud-adapters

parents c3350080 2763b3e2

README.md

+1 −1
Original line number Diff line number Diff line
@@ -40,7 +40,7 @@ Open source SDK to abstract CAMARA/GSMA Transformation Functions (TFs) for Edge 


| API Name                  | Version |

|---------------------------|---------|

| Edge Application Management | [v0.9.3-wip](https://raw.githubusercontent.com/camaraproject/EdgeCloud/main/code/API_definitions/Edge-Application-Management.yaml) |

| Edge Application Management | [v0.9.3-wip (commit: 79aa595)](https://github.com/camaraproject/EdgeCloud/blob/79aa5951d64391422ea5b861ab617a9540386092/code/API_definitions/Edge-Application-Management.yaml) |

| Quality-on-Demand         | [v1.0.0](https://raw.githubusercontent.com/camaraproject/QualityOnDemand/refs/tags/r2.2/code/API_definitions/quality-on-demand.yaml) |

| Location Retrieval        | [v0.4.0](https://raw.githubusercontent.com/camaraproject/DeviceLocation/refs/tags/r2.2/code/API_definitions/location-retrieval.yaml) |

| Traffic Influence         | [v0.8.1](https://raw.githubusercontent.com/camaraproject/EdgeCloud/v0.8.1/code/API_definitions/Traffic_Influence.yaml) |

pyproject.toml

+1 −1
Original line number Diff line number Diff line
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" 


[project]

name = "sunrise6g-opensdk"

version = "1.0.4"

version = "1.0.5"

description = "Open source SDK to abstract CAMARA/GSMA Transformation Functions (TFs) for Edge Cloud platforms, 5G network cores and Open RAN solutions."

keywords = [

  "Federation",

src/sunrise6g_opensdk/edgecloud/adapters/i2edge/client.py

+9 −6
Original line number Diff line number Diff line
@@ -112,7 +112,6 @@ class EdgeApplicationManager(EdgeCloudManagementInterface): 
    # ------------------------------------------------------------------------

    # Artefact Management (i2Edge-Specific, Non-CAMARA)

    # ------------------------------------------------------------------------

    # All artefact methods now return Response objects for API consistency



    def create_artefact(

        self,
@@ -383,7 +382,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface): 
                        name=app_metadata.get("appName", ""),

                        version=app_metadata.get("version", ""),

                        appProvider=profile_data.get("appProviderId", ""),

                        # Hardcoding mandatory fields that don't exist in i2Edge

                        # Hardcoding mandatory fields that doesn't exist in i2Edge

                        packageType="CONTAINER",

                        appRepo={"type": "PUBLICREPO", "imagePath": "not-available"},

                        requiredResources={
@@ -465,18 +464,22 @@ class EdgeApplicationManager(EdgeCloudManagementInterface): 
                appId=camara_schemas.AppId(appId),

                appInstanceId=camara_schemas.AppInstanceId(app_instance_id),

                appProvider=camara_schemas.AppProvider(appProviderId),

                status=camara_schemas.Status.instantiating,  # 202 means deployment is in progress

                status=camara_schemas.Status.instantiating,

                edgeCloudZoneId=camara_schemas.EdgeCloudZoneId(zone_id),

            )



            # CAMARA spec requires appInstances array wrapper

            camara_response = {"appInstances": [app_instance_info.model_dump(mode="json")]}

            camara_response = app_instance_info.model_dump(mode="json")



            # Add mandatory Location header

            location_url = f"/appinstances/{app_instance_id}"

            camara_headers = {"Content-Type": "application/json", "Location": location_url}



            log.info("App deployment request submitted successfully")

            return build_custom_http_response(

                status_code=i2edge_response.status_code,

                content=camara_response,

                headers={"Content-Type": "application/json"},

                headers=camara_headers,

                encoding="utf-8",

                url=i2edge_response.url,

                request=i2edge_response.request,
@@ -544,7 +547,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface): 
                            ),

                            status=camara_schemas.Status(

                                camara_status

                            ),  # Map the i2Edge "DEPLOYED" status to the CAMARA "ready" status for consistency with CAMARA specifications.

                            ),  # FIX: Map DEPLOYED -> ready

                            edgeCloudZoneId=camara_schemas.EdgeCloudZoneId(

                                zone_id

                            ),  # FIX: Extract from nodeSelector

tests/edgecloud/test_e2e.py

+52 −43
Original line number Diff line number Diff line
@@ -101,14 +101,27 @@ def test_config_camara_compliance(edgecloud_client): 


@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:

            camara_schemas.EdgeCloudZone(**zone)

            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:
@@ -144,8 +157,10 @@ def test_onboard_app(edgecloud_client): 


        payload = response.json()

        assert isinstance(payload, dict)

        assert "appId" in payload

        camara_schemas.AppId(root=payload["appId"])



        # 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}")
@@ -161,22 +176,20 @@ def app_instance_id(edgecloud_client): 
        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)



        # All CAMARA-compliant adapters should return 202 for async deployment

        assert response.status_code == 202

        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)



        # CAMARA spec: response contains appInstances array

        assert "appInstances" in response_data

        assert isinstance(response_data["appInstances"], list)

        assert len(response_data["appInstances"]) > 0



        # Extract appInstanceId from first instance

        app_instance_id = response_data["appInstances"][0].get("appInstanceId")

        # Extract appInstanceId from the validated object

        app_instance_id = instance_info.appInstanceId.root



        assert app_instance_id is not None

        yield app_instance_id
@@ -208,11 +221,9 @@ def test_get_onboarded_app(edgecloud_client): 
        assert isinstance(app_data, dict)

        assert "appManifest" in app_data



        app_manifest = app_data["appManifest"]

        assert app_manifest["appId"] == app_id

        assert "name" in app_manifest

        assert "version" in app_manifest

        assert "appProvider" in app_manifest

        # 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}")
@@ -221,6 +232,7 @@ def test_get_onboarded_app(edgecloud_client): 
@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)
@@ -229,16 +241,16 @@ def test_get_all_onboarded_apps(edgecloud_client): 
        apps_data = response.json()

        assert isinstance(apps_data, list)



        # Verify each app has required CAMARA fields

        for app_manifest in apps_data:

            assert "appId" in app_manifest

            assert "name" in app_manifest

            assert "version" in app_manifest

            assert "appProvider" in app_manifest

            assert "packageType" in app_manifest

            assert "appRepo" in app_manifest

            assert "requiredResources" in app_manifest

            assert "componentSpec" in app_manifest

        # 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}")
@@ -257,14 +269,13 @@ def test_get_all_deployed_apps(edgecloud_client): 
        assert "appInstances" in instances_data

        assert isinstance(instances_data["appInstances"], list)



        # Verify each instance has required CAMARA fields

        for instance in instances_data["appInstances"]:

            assert "name" in instance

            assert "appId" in instance

            assert "appInstanceId" in instance

            assert "appProvider" in instance

            assert "status" in instance

            assert "edgeCloudZoneId" in instance

        # 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}")
@@ -282,13 +293,11 @@ def test_get_deployed_app(edgecloud_client, app_instance_id): 
        assert isinstance(instance_data, dict)

        assert "appInstance" in instance_data



        instance = instance_data["appInstance"]

        assert instance["appInstanceId"] == app_instance_id

        assert "name" in instance

        assert "appId" in instance

        assert "appProvider" in instance

        assert "status" in instance

        assert "edgeCloudZoneId" in instance

        # 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}")