fix: update AppInstance status mapping and add utility function for state conversion (cf39aa51) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/mappers/edge_application_mapper.py

+16 −2

Original line number	Diff line number	Diff line
		@@ -62,7 +62,7 @@ from open_exposure_gateway.domain.edge_application_management import (
		SRMTerminatePayload,
		SRMTopologyConstraints,
		)
		from open_exposure_gateway.domain.models import AppInstance
		from open_exposure_gateway.domain.models import AppInstance, AppInstanceState

		_PACKAGE_TYPE_TO_RUNTIME_KIND: dict[str, str] = {
		"HELM": "helm",
		@@ -95,9 +95,23 @@ _SRM_STATE_TO_APP_INSTANCE_STATUS: dict[str, AppInstanceStatus] = {
		"degraded": AppInstanceStatus.FAILED,
		"failed": AppInstanceStatus.FAILED,
		"terminating": AppInstanceStatus.TERMINATING,
		"terminated": AppInstanceStatus.TERMINATING,
		"terminated": AppInstanceStatus.UNKNOWN,
		}

		# CAMARA's AppInstanceStatus has no `terminated`; the internal terminal state
		# surfaces as `unknown` (app-instance-flow.md, AppInstanceStatus Mapping).
		_APP_INSTANCE_STATE_TO_STATUS: dict[AppInstanceState, AppInstanceStatus] = {
		AppInstanceState.INSTANTIATING: AppInstanceStatus.INSTANTIATING,
		AppInstanceState.READY: AppInstanceStatus.READY,
		AppInstanceState.FAILED: AppInstanceStatus.FAILED,
		AppInstanceState.TERMINATING: AppInstanceStatus.TERMINATING,
		AppInstanceState.TERMINATED: AppInstanceStatus.UNKNOWN,
		}


		def to_app_instance_status(state: AppInstanceState) -> AppInstanceStatus:
		return _APP_INSTANCE_STATE_TO_STATUS[state]


		def _parse_container_cpu_cores(value: str) -> float:
		if value.endswith("m"):

src/open_exposure_gateway/application/services/edge_application_management_service.py

+2 −1

Original line number	Diff line number	Diff line
		@@ -27,6 +27,7 @@ from open_exposure_gateway.application.mappers.edge_application_mapper import (
		build_edge_cloud_zone,
		build_submitted_app,
		build_terminate_instance_command,
		to_app_instance_status,
		)
		from open_exposure_gateway.core.exceptions import (
		AbortedException,
		@@ -322,7 +323,7 @@ class EdgeApplicationManagementService:
		name=request.name,
		appId=request.appId,
		appProvider=app_provider_id,
		status=AppInstanceStatus(existing_instance.state),
		status=to_app_instance_status(existing_instance.state),
		edgeCloudZoneId=request.edgeCloudZoneId,
		)

tests/unit/test_eam_mapper.py

+29 −0

Original line number	Diff line number	Diff line
		from uuid import UUID

		import pytest

		from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas import (
		AppInstanceStatus,
		ApplicationResources,
		@@ -23,6 +25,7 @@ from open_exposure_gateway.application.mappers.edge_application_mapper import (
		build_catalog_payload,
		build_deploy_command,
		build_edge_cloud_zone,
		to_app_instance_status,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		SRMAccelerator,
		@@ -348,6 +351,10 @@ class TestBuildAppInstanceInfo:
		result = build_app_instance_info(self._make_instance(state="terminating"), APP_ID)
		assert result.status == AppInstanceStatus.TERMINATING

		def test_state_mapping_terminated_to_unknown(self) -> None:
		result = build_app_instance_info(self._make_instance(state="terminated"), APP_ID)
		assert result.status == AppInstanceStatus.UNKNOWN

		def test_unknown_state_defaults_to_unknown(self) -> None:
		result = build_app_instance_info(self._make_instance(state="exotic"), APP_ID)
		assert result.status == AppInstanceStatus.UNKNOWN
		@@ -385,6 +392,28 @@ class TestBuildAppInstanceInfo:
		assert result.name == str(INSTANCE_ID)


		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)."""

		@pytest.mark.parametrize(
		("state", "expected"),
		[
		(AppInstanceState.INSTANTIATING, AppInstanceStatus.INSTANTIATING),
		(AppInstanceState.READY, AppInstanceStatus.READY),
		(AppInstanceState.FAILED, AppInstanceStatus.FAILED),
		(AppInstanceState.TERMINATING, AppInstanceStatus.TERMINATING),
		(AppInstanceState.TERMINATED, AppInstanceStatus.UNKNOWN),
		],
		)
		def test_maps_every_state(self, state: AppInstanceState, expected: AppInstanceStatus) -> None:
		assert to_app_instance_status(state) == expected

		def test_covers_every_state(self) -> None:
		for state in AppInstanceState:
		to_app_instance_status(state)


		class TestBuildAppRegistrationTranslation:
		def test_helm_translation_fields(self) -> None:
		manifest = _make_helm_manifest()

tests/unit/test_eam_service.py

+28 −0

Original line number	Diff line number	Diff line
		@@ -830,6 +830,34 @@ class TestCreateAppInstance:
		)
		assert second.status == AppInstanceStatus.READY

		async def test_replay_of_terminated_instance_returns_unknown_status(
		self,
		service: EdgeApplicationManagementService,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		"""`terminated` has no CAMARA AppInstanceStatus counterpart, so the
		replay path must map it rather than construct the enum from the state."""
		first = await service.create_app_instance(
		request=self._make_request(),
		tenant_id="tenant-1",
		app_provider_id="provider-1",
		idempotency_key="retry-key-1",
		)
		stored = await app_instance_repo.get_by_id(first.appInstanceId)
		assert stored is not None
		await app_instance_repo.save(
		stored.model_copy(update={"state": AppInstanceState.TERMINATED})
		)

		second = await service.create_app_instance(
		request=self._make_request(),
		tenant_id="tenant-1",
		app_provider_id="provider-1",
		idempotency_key="retry-key-1",
		)
		assert second.appInstanceId == first.appInstanceId
		assert second.status == AppInstanceStatus.UNKNOWN

		async def test_different_keys_create_separate_operations(
		self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository
		) -> None:

Admin message