feat: add build_deploy_targets and build_multi_zone_deploy_command mappers (082fb9e0) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/mappers/edge_application_mapper.py

+42 −0

Original line number	Diff line number	Diff line
		@@ -11,6 +11,7 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas i
		ComponentEndpointInfo,
		ComponentSpecItem,
		ContainerResources,
		CreateAppDeploymentRequest,
		CreateAppInstanceRequest,
		DockerComposeResources,
		EdgeCloudZone,
		@@ -729,6 +730,47 @@ def build_deploy_command(
		)


		def build_deploy_targets(
		request: CreateAppDeploymentRequest, app_instance_ids: list[UUID]
		) -> list[SRMDeployTarget]:
		"""One targets[] entry per edgeCloudZones[i]; kubernetesClusterRefs[i] pairs
		positionally (OQ-D1, oeg/app-deployment-flow.md). Zones beyond the length of
		kubernetesClusterRefs get no domain pin (SRM auto-places)."""
		cluster_refs = request.kubernetesClusterRefs or []
		if len(cluster_refs) > len(request.edgeCloudZones):
		raise ValueError("kubernetesClusterRefs must not be longer than edgeCloudZones")
		return [
		SRMDeployTarget(
		app_instance_id=str(app_instance_id),
		zone_id=str(zone_id),
		domain_id=str(cluster_refs[i]) if i < len(cluster_refs) else None,
		)
		for i, (zone_id, app_instance_id) in enumerate(
		zip(request.edgeCloudZones, app_instance_ids, strict=True)
		)
		]


		def build_multi_zone_deploy_command(
		app_registration_id: UUID,
		targets: list[SRMDeployTarget],
		name: str,
		operation_id: UUID,
		correlation_id: str,
		requested_at: str,
		app_provider_id: str,
		) -> SRMDeployCommand:
		return SRMDeployCommand(
		operation_id=str(operation_id),
		correlation_id=correlation_id,
		requested_at=requested_at,
		app_provider_id=app_provider_id,
		service_specification_id=str(app_registration_id),
		targets=targets,
		deploy=SRMDeployPayload(instance_name=name),
		)


		def build_submitted_app(translation: AppRegistrationTranslation) -> SubmittedApp:
		return SubmittedApp(appId=translation.app_id)

tests/unit/test_eam_mapper.py

+103 −0

Original line number	Diff line number	Diff line
		@@ -10,6 +10,7 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas i
		AppRepo,
		ComponentSpecItem,
		ContainerResources,
		CreateAppDeploymentRequest,
		CreateAppInstanceRequest,
		DockerComposeResources,
		EdgeCloudZoneStatus,
		@@ -25,7 +26,9 @@ from open_exposure_gateway.application.mappers.edge_application_mapper import (
		build_app_registration_translation,
		build_catalog_payload,
		build_deploy_command,
		build_deploy_targets,
		build_edge_cloud_zone,
		build_multi_zone_deploy_command,
		to_app_instance_status,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		@@ -928,6 +931,106 @@ class TestBuildDeployCommand:
		assert cmd.deploy.placement_constraints == {}


		class TestBuildDeployTargets:
		ZONE_A = UUID("aaaaaaaa-0000-4000-8000-00000000000a")
		ZONE_B = UUID("aaaaaaaa-0000-4000-8000-00000000000b")
		INSTANCE_A = UUID("cccccccc-0000-4000-8000-00000000000a")
		INSTANCE_B = UUID("cccccccc-0000-4000-8000-00000000000b")
		CLUSTER_A = UUID("dddddddd-0000-4000-8000-00000000000a")

		def test_one_target_per_zone(self) -> None:
		request = CreateAppDeploymentRequest(
		appDeploymentName="video_analytics_eu",
		appId=APP_ID,
		edgeCloudZones=[self.ZONE_A, self.ZONE_B],
		)
		targets = build_deploy_targets(request, [self.INSTANCE_A, self.INSTANCE_B])
		assert len(targets) == 2
		assert targets[0].zone_id == str(self.ZONE_A)
		assert targets[0].app_instance_id == str(self.INSTANCE_A)
		assert targets[1].zone_id == str(self.ZONE_B)
		assert targets[1].app_instance_id == str(self.INSTANCE_B)

		def test_cluster_refs_pair_positionally(self) -> None:
		# kubernetesClusterRefs shorter than edgeCloudZones: only the leading
		# zones get a domain pin, the rest are left for SRM to auto-place (OQ-D1).
		request = CreateAppDeploymentRequest(
		appDeploymentName="video_analytics_eu",
		appId=APP_ID,
		edgeCloudZones=[self.ZONE_A, self.ZONE_B],
		kubernetesClusterRefs=[self.CLUSTER_A],
		)
		targets = build_deploy_targets(request, [self.INSTANCE_A, self.INSTANCE_B])
		assert targets[0].domain_id == str(self.CLUSTER_A)
		assert targets[1].domain_id is None

		def test_no_cluster_refs_gives_no_domain_pins(self) -> None:
		request = CreateAppDeploymentRequest(
		appDeploymentName="video_analytics_eu",
		appId=APP_ID,
		edgeCloudZones=[self.ZONE_A],
		)
		targets = build_deploy_targets(request, [self.INSTANCE_A])
		assert targets[0].domain_id is None

		def test_cluster_refs_longer_than_zones_raises(self) -> None:
		request = CreateAppDeploymentRequest(
		appDeploymentName="video_analytics_eu",
		appId=APP_ID,
		edgeCloudZones=[self.ZONE_A],
		kubernetesClusterRefs=[self.CLUSTER_A, UUID(int=0)],
		)
		with pytest.raises(ValueError, match="kubernetesClusterRefs"):
		build_deploy_targets(request, [self.INSTANCE_A])


		class TestBuildMultiZoneDeployCommand:
		def test_fields_set_correctly(self) -> None:
		request = CreateAppDeploymentRequest(
		appDeploymentName="video_analytics_eu",
		appId=APP_ID,
		edgeCloudZones=[ZONE_ID],
		)
		targets = build_deploy_targets(request, [INSTANCE_ID])
		cmd = build_multi_zone_deploy_command(
		app_registration_id=APP_REGISTRATION_ID,
		targets=targets,
		name="video_analytics_eu",
		operation_id=OP_ID,
		correlation_id="corr-456",
		requested_at="2026-07-04T10:00:00Z",
		app_provider_id="provider-1",
		)
		assert cmd.service_specification_id == str(APP_REGISTRATION_ID)
		assert cmd.targets == targets
		assert cmd.operation_id == str(OP_ID)
		assert cmd.correlation_id == "corr-456"
		assert cmd.app_provider_id == "provider-1"
		assert cmd.deploy.instance_name == "video_analytics_eu"
		assert cmd.source == "nbi_camara"
		assert cmd.deploy.placement_constraints == {}

		def test_carries_n_targets_for_multi_zone(self) -> None:
		zone_a, zone_b = UUID(int=1), UUID(int=2)
		instance_a, instance_b = UUID(int=3), UUID(int=4)
		request = CreateAppDeploymentRequest(
		appDeploymentName="video_analytics_eu",
		appId=APP_ID,
		edgeCloudZones=[zone_a, zone_b],
		)
		targets = build_deploy_targets(request, [instance_a, instance_b])
		cmd = build_multi_zone_deploy_command(
		app_registration_id=APP_REGISTRATION_ID,
		targets=targets,
		name="video_analytics_eu",
		operation_id=OP_ID,
		correlation_id="corr-456",
		requested_at="2026-07-04T10:00:00Z",
		app_provider_id="provider-1",
		)
		assert len(cmd.targets) == 2


		class TestBuildAppInstanceStatusChangeEvent:
		def _make_app_instance(self, state: AppInstanceState) -> AppInstance:
		return AppInstance(