Commit 7d7297a0 authored by Adrian Pino's avatar Adrian Pino
Browse files

Refactors zone data transformation to CAMARA format 

Introduces a dedicated function to transform i2Edge zone data into the CAMARA EdgeCloudZone format, improving code readability and maintainability.

Improves error handling for invalid CAMARA manifests by providing more detailed information about validation failures.

Increases the wait time in the end-to-end tests to avoid potential timing issues.
parent b38e605c

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

+25 −13
Original line number Diff line number Diff line
@@ -43,6 +43,25 @@ class EdgeApplicationManager(EdgeCloudManagementInterface): 
        self.content_type_gsma = "application/json"

        self.encoding_gsma = "utf-8"



    def _transform_to_camara_zone(self, zone_data: dict) -> camara_schemas.EdgeCloudZone:

        """

        Transform i2Edge zone data to CAMARA EdgeCloudZone format.



        :param zone_data: Raw zone data from i2Edge API

        :return: CAMARA-compliant EdgeCloudZone object

        """

        return camara_schemas.EdgeCloudZone(

            edgeCloudZoneId=camara_schemas.EdgeCloudZoneId(zone_data.get("zoneId", "unknown")),

            edgeCloudZoneName=camara_schemas.EdgeCloudZoneName(

                zone_data.get("nodeName", "unknown")

            ),

            edgeCloudProvider=camara_schemas.EdgeCloudProvider("i2edge"),

            edgeCloudRegion=camara_schemas.EdgeCloudRegion(

                zone_data.get("geographyDetails", "unknown")

            ),

            edgeCloudZoneStatus=camara_schemas.EdgeCloudZoneStatus.unknown,

        )



    # ########################################################################

    # CAMARA EDGE CLOUD MANAGEMENT API

    # ########################################################################
@@ -71,17 +90,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface): 
                i2edge_response = response.json()

                log.info("Availability zones retrieved successfully")

                # Normalise to CAMARA format

                camara_response = []

                for z in i2edge_response:

                    zone = camara_schemas.EdgeCloudZone(

                        # edgeCloudZoneId = camara_schemas.EdgeCloudZoneId(z["zoneId"]),

                        edgeCloudZoneId=camara_schemas.EdgeCloudZoneId(z["zoneId"]),

                        edgeCloudZoneName=camara_schemas.EdgeCloudZoneName(z["nodeName"]),

                        edgeCloudProvider=camara_schemas.EdgeCloudProvider("i2edge"),

                        edgeCloudRegion=camara_schemas.EdgeCloudRegion(z["geographyDetails"]),

                        edgeCloudZoneStatus=camara_schemas.EdgeCloudZoneStatus.unknown,

                    )

                    camara_response.append(zone)

                camara_response = [self._transform_to_camara_zone(z) for z in i2edge_response]

                # Wrap into a Response object

                return build_custom_http_response(

                    status_code=response.status_code,
@@ -267,8 +276,11 @@ class EdgeApplicationManager(EdgeCloudManagementInterface): 
                i2edge_response.raise_for_status()

        # TODO: Implement CAMARA-compliant error handling for failed onboarding responses

        except ValidationError as e:

            log.error(f"Invalid CAMARA manifest: {e}")

            raise ValueError(f"Invalid CAMARA manifest: {e}")

            error_details = "; ".join(

                [f"Field '{err['loc'][0]}': {err['msg']}" for err in e.errors()]

            )

            log.error(f"Invalid CAMARA manifest: {error_details}")

            raise ValueError(f"Invalid CAMARA manifest: {error_details}")

        except I2EdgeError as e:

            log.error(f"Failed to onboard app to i2Edge: {e}")

            raise

tests/edgecloud/test_e2e.py

+2 −2
Original line number Diff line number Diff line
@@ -190,8 +190,8 @@ def test_deploy_app(app_instance_id): 




@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)

def test_timer_wait_10_seconds(edgecloud_client):

    time.sleep(10)

def test_timer_wait_30_seconds(edgecloud_client):

    time.sleep(30)





@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)