feat: add get_app_deployments to EdgeApplicationManagementService (0d887242) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/edge_application_management_service.py

+67 −0

Original line number	Diff line number	Diff line
		@@ -8,6 +8,7 @@ from pydantic import BaseModel
		from open_exposure_gateway.adapters.errors import DuplicateAppRegistrationError
		from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas import (
		AppDeploymentId,
		AppDeploymentInfo,
		AppInstanceInfo,
		AppInstanceStatus,
		AppManifest,
		@@ -19,6 +20,7 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas i
		SubmittedApp,
		)
		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,
		@@ -549,6 +551,71 @@ class EdgeApplicationManagementService:
		)
		return AppDeploymentId(appDeploymentId=app_deployment_id)

		async def get_app_deployments(
		self,
		app_id: Optional[UUID] = None,
		app_deployment_id: Optional[UUID] = None,
		x_correlator: Optional[str] = None,
		) -> list[AppDeploymentInfo]:
		if self._app_registration_repo is None:
		raise RuntimeError("AppRegistrationRepository is not available")
		if self._app_deployment_repo is None or self._app_instance_repo is None:
		raise RuntimeError("AppDeployment/AppInstance repositories are not available")

		resolved_app_registration_id: Optional[UUID] = None
		if app_id is not None:
		app_registration = await self._app_registration_repo.get_by_app_id(app_id)
		if app_registration is None:
		return []
		resolved_app_registration_id = app_registration.app_registration_id

		deployments: list[AppDeployment]
		if app_deployment_id is not None:
		deployment = await self._app_deployment_repo.get_by_id(app_deployment_id)
		if deployment is None:
		return []
		if (
		resolved_app_registration_id is not None
		and deployment.app_registration_id != resolved_app_registration_id
		):
		return []
		deployments = [deployment]
		elif resolved_app_registration_id is not None:
		deployments = await self._app_deployment_repo.list_by_app_registration_id(
		resolved_app_registration_id
		)
		else:
		deployments = await self._app_deployment_repo.list_all()

		result: list[AppDeploymentInfo] = []
		app_id_cache: dict[UUID, Optional[UUID]] = {}
		if app_id is not None and resolved_app_registration_id is not None:
		app_id_cache[resolved_app_registration_id] = app_id
		for deployment in deployments:
		if deployment.app_registration_id not in app_id_cache:
		owner = await self._app_registration_repo.get_by_id(deployment.app_registration_id)
		app_id_cache[deployment.app_registration_id] = owner.app_id if owner else None
		deployment_app_id = app_id_cache[deployment.app_registration_id]
		if deployment_app_id is None:
		logger.warning(
		"app_deployment_listed_for_unresolvable_app_registration",
		app_deployment_id=str(deployment.app_deployment_id),
		app_registration_id=str(deployment.app_registration_id),
		)
		continue

		app_instances = await self._app_instance_repo.list_by_app_deployment_id(
		deployment.app_deployment_id
		)
		result.append(
		build_app_deployment_info(
		deployment,
		deployment_app_id,
		[instance.app_instance_id for instance in app_instances],
		)
		)
		return result

		async def get_app_instances(
		self,
		app_id: Optional[UUID] = None,

tests/unit/test_eam_service.py

+140 −0

Original line number	Diff line number	Diff line
		@@ -76,6 +76,7 @@ APP_ID = UUID("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb")
		ZONE_ID = UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
		INSTANCE_ID = UUID("cccccccc-cccc-cccc-cccc-cccccccccccc")
		APP_REGISTRATION_ID = UUID("ffffffff-ffff-ffff-ffff-ffffffffffff")
		DEPLOYMENT_ID = UUID("11111111-1111-1111-1111-111111111111")


		async def _seed_registration(
		@@ -1450,6 +1451,145 @@ class TestCreateAppDeploymentUnregisteredApp:
		)


		class TestGetAppDeployments:
		async def _seed_deployment(
		self,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_registration_id: UUID = APP_REGISTRATION_ID,
		app_deployment_id: UUID = DEPLOYMENT_ID,
		) -> AppDeployment:
		return await app_deployment_repo.save(
		AppDeployment(
		app_deployment_id=app_deployment_id,
		operation_id=uuid4(),
		app_registration_id=app_registration_id,
		app_deployment_name="video_analytics_eu",
		edge_cloud_zones=[ZONE_ID],
		state=AppDeploymentState.READY,
		)
		)

		async def test_no_filters_returns_all_deployments(
		self,
		service: EdgeApplicationManagementService,
		app_registration_repo: FakeAppRegistrationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await _seed_registration(app_registration_repo)
		await self._seed_deployment(app_deployment_repo)

		result = await service.get_app_deployments()
		assert len(result) == 1
		assert result[0].appDeploymentId == DEPLOYMENT_ID
		assert result[0].appId == APP_ID

		async def test_no_deployments_returns_empty(
		self, service: EdgeApplicationManagementService
		) -> None:
		assert await service.get_app_deployments() == []

		async def test_app_id_filter_returns_only_that_apps_deployments(
		self,
		service: EdgeApplicationManagementService,
		app_registration_repo: FakeAppRegistrationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await _seed_registration(app_registration_repo)
		await self._seed_deployment(app_deployment_repo)

		other_app_registration_id = uuid4()
		await _seed_registration(
		app_registration_repo, app_id=uuid4(), app_registration_id=other_app_registration_id
		)
		await self._seed_deployment(
		app_deployment_repo,
		app_registration_id=other_app_registration_id,
		app_deployment_id=uuid4(),
		)

		result = await service.get_app_deployments(app_id=APP_ID)
		assert len(result) == 1
		assert result[0].appDeploymentId == DEPLOYMENT_ID

		async def test_unregistered_app_id_filter_returns_empty(
		self, service: EdgeApplicationManagementService
		) -> None:
		assert await service.get_app_deployments(app_id=APP_ID) == []

		async def test_app_deployment_id_filter_returns_single_result(
		self,
		service: EdgeApplicationManagementService,
		app_registration_repo: FakeAppRegistrationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await _seed_registration(app_registration_repo)
		await self._seed_deployment(app_deployment_repo)
		await self._seed_deployment(app_deployment_repo, app_deployment_id=uuid4())

		result = await service.get_app_deployments(app_deployment_id=DEPLOYMENT_ID)
		assert len(result) == 1
		assert result[0].appDeploymentId == DEPLOYMENT_ID

		async def test_unknown_app_deployment_id_returns_empty(
		self, service: EdgeApplicationManagementService
		) -> None:
		assert await service.get_app_deployments(app_deployment_id=uuid4()) == []

		async def test_consistent_app_id_and_app_deployment_id_returns_result(
		self,
		service: EdgeApplicationManagementService,
		app_registration_repo: FakeAppRegistrationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await _seed_registration(app_registration_repo)
		await self._seed_deployment(app_deployment_repo)

		result = await service.get_app_deployments(app_id=APP_ID, app_deployment_id=DEPLOYMENT_ID)
		assert len(result) == 1

		async def test_inconsistent_app_id_and_app_deployment_id_returns_empty(
		self,
		service: EdgeApplicationManagementService,
		app_registration_repo: FakeAppRegistrationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await _seed_registration(app_registration_repo)
		await self._seed_deployment(app_deployment_repo)

		other_app_id = uuid4()
		await _seed_registration(
		app_registration_repo, app_id=other_app_id, app_registration_id=uuid4()
		)

		result = await service.get_app_deployments(
		app_id=other_app_id, app_deployment_id=DEPLOYMENT_ID
		)
		assert result == []

		async def test_includes_app_instance_ids(
		self,
		service: EdgeApplicationManagementService,
		app_registration_repo: FakeAppRegistrationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await _seed_registration(app_registration_repo)
		await self._seed_deployment(app_deployment_repo)
		await app_instance_repo.save(
		AppInstance(
		app_instance_id=INSTANCE_ID,
		operation_id=uuid4(),
		app_registration_id=APP_REGISTRATION_ID,
		edge_cloud_zone_id=ZONE_ID,
		state=AppInstanceState.READY,
		app_deployment_id=DEPLOYMENT_ID,
		)
		)

		result = await service.get_app_deployments()
		assert result[0].appInstances == [INSTANCE_ID]


		class TestGetAppInstances:
		async def test_returns_mapped_instances(
		self,