Commit a5ecfdde authored by George Papathanail's avatar George Papathanail
Browse files

feat: add build_app_deployment_info mapper

parent 98fffbea

src/open_exposure_gateway/application/mappers/edge_application_mapper.py

+14 −1
Original line number Diff line number Diff line
@@ -5,6 +5,7 @@ from uuid import UUID, uuid4 


from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas import (

    AccessEndpoint,

    AppDeploymentInfo,

    AppInstanceInfo,

    AppInstanceStatus,

    AppManifest,
@@ -72,7 +73,7 @@ from open_exposure_gateway.domain.edge_application_management import ( 
    SRMZone,

    StorageRequest,

)

from open_exposure_gateway.domain.models import AppInstance, AppInstanceState

from open_exposure_gateway.domain.models import AppDeployment, AppInstance, AppInstanceState



_PACKAGE_TYPE_TO_RUNTIME_KIND: dict[str, str] = {

    "HELM": "helm",
@@ -400,6 +401,18 @@ def build_app_instance_info( 
    )





def build_app_deployment_info(

    deployment: AppDeployment, app_id: UUID, app_instance_ids: list[UUID]

) -> AppDeploymentInfo:

    return AppDeploymentInfo(

        appDeploymentName=deployment.app_deployment_name,

        appDeploymentId=deployment.app_deployment_id,

        appId=app_id,

        edgeCloudZones=deployment.edge_cloud_zones,

        appInstances=app_instance_ids,

    )





def build_app_registration_translation(

    manifest: AppManifest,

    app_id: UUID,

tests/unit/test_eam_mapper.py

+41 −1
Original line number Diff line number Diff line
@@ -19,6 +19,7 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas i 
    VmResources,

)

from open_exposure_gateway.application.mappers.edge_application_mapper import (

    build_app_deployment_info,

    build_app_deployment_translation,

    build_app_instance_info,

    build_app_instance_status_change_event,
@@ -52,13 +53,19 @@ from open_exposure_gateway.domain.edge_application_management import ( 
    SRMZoneLocation,

    SRMZoneMetadata,

)

from open_exposure_gateway.domain.models import AppInstance, AppInstanceState

from open_exposure_gateway.domain.models import (

    AppDeployment,

    AppDeploymentState,

    AppInstance,

    AppInstanceState,

)



APP_ID = UUID("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb")

ZONE_ID = UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")

INSTANCE_ID = UUID("cccccccc-cccc-cccc-cccc-cccccccccccc")

OP_ID = UUID("dddddddd-dddd-dddd-dddd-dddddddddddd")

APP_REGISTRATION_ID = UUID("ffffffff-ffff-ffff-ffff-ffffffffffff")

DEPLOYMENT_ID = UUID("eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee")





def _make_srm_catalog(
@@ -448,6 +455,39 @@ class TestBuildAppInstanceInfo: 
        assert result.name == str(INSTANCE_ID)





class TestBuildAppDeploymentInfo:

    def _make_deployment(self) -> AppDeployment:

        return AppDeployment(

            app_deployment_id=DEPLOYMENT_ID,

            operation_id=OP_ID,

            app_registration_id=APP_REGISTRATION_ID,

            app_deployment_name="video_analytics_eu",

            edge_cloud_zones=[ZONE_ID],

            state=AppDeploymentState.READY,

        )



    def test_maps_ids_and_name(self) -> None:

        result = build_app_deployment_info(self._make_deployment(), APP_ID, [INSTANCE_ID])

        assert result.appDeploymentId == DEPLOYMENT_ID

        assert result.appDeploymentName == "video_analytics_eu"

        assert result.appId == APP_ID



    def test_maps_edge_cloud_zones(self) -> None:

        result = build_app_deployment_info(self._make_deployment(), APP_ID, [INSTANCE_ID])

        assert result.edgeCloudZones == [ZONE_ID]



    def test_maps_app_instance_ids(self) -> None:

        other_instance_id = UUID("11111111-1111-1111-1111-111111111111")

        result = build_app_deployment_info(

            self._make_deployment(), APP_ID, [INSTANCE_ID, other_instance_id]

        )

        assert result.appInstances == [INSTANCE_ID, other_instance_id]



    def test_empty_app_instances_gives_empty_list(self) -> None:

        result = build_app_deployment_info(self._make_deployment(), APP_ID, [])

        assert result.appInstances == []





class TestToAppInstanceStatus:

    """Every internal state must map to a CAMARA-legal status; `terminated` has

    no CAMARA counterpart and surfaces as `unknown` (app-instance-flow.md)."""