Merge branch 'fix/instance-list-lifecycle' into 'develop' (10149419) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/database/repos/app_instances.py

+13 −1

Changes for src/open_exposure_gateway/adapters/database/repos/app_instances.py: 13 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -4,7 +4,7 @@ from sqlalchemy import select
		from sqlalchemy.ext.asyncio import AsyncSession

		from open_exposure_gateway.adapters.database.mappers import AppInstanceMapper
		from open_exposure_gateway.adapters.database.sql import AppInstanceRow
		from open_exposure_gateway.adapters.database.sql import AppInstanceRow, OperationRow
		from open_exposure_gateway.domain.models import AppInstance, AppInstanceState
		from open_exposure_gateway.ports.database.instances import AppInstanceRepository

		@@ -20,6 +20,18 @@ class SqlAppInstanceRepository(AppInstanceRepository):
		row = await self._session.scalar(stmt)
		return AppInstanceMapper.to_domain(row) if row is not None else None

		async def list_failed_by_provider(self, app_provider_id: str) -> list[AppInstance]:
		stmt = (
		select(AppInstanceRow)
		.join(OperationRow, AppInstanceRow.operation_id == OperationRow.operation_id)
		.where(
		OperationRow.app_provider_id == app_provider_id,
		AppInstanceRow.state == AppInstanceState.FAILED,
		)
		)
		rows = await self._session.scalars(stmt)
		return [AppInstanceMapper.to_domain(row) for row in rows]

		async def get_by_operation_id(self, operation_id: UUID) -> AppInstance \| None:
		stmt = select(AppInstanceRow).where(AppInstanceRow.operation_id == operation_id)
		row = await self._session.scalar(stmt)

src/open_exposure_gateway/application/services/edge_application_management_service.py

+61 −0

Changes for src/open_exposure_gateway/application/services/edge_application_management_service.py: 61 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -369,6 +369,7 @@ class EdgeApplicationManagementService:
		catalogs = await self.srm_client.get_apps(
		app_provider_id=app_provider_id, x_correlator=x_correlator
		)
		catalogs = [catalog for catalog in catalogs if not await self._deleted_here(catalog)]
		try:
		return [build_app_manifest(catalog) for catalog in catalogs]
		except (ValueError, TypeError, IndexError) as exc:
		@@ -967,6 +968,10 @@ class EdgeApplicationManagementService:
		result: list[AppInstanceInfo] = []
		registration_cache: dict[UUID, Optional[AppRegistration]] = {}
		for instance in instances:
		if self._app_instance_repo is not None:
		saved = await self._app_instance_repo.get_by_id(UUID(instance.service_instance_id))
		if saved is not None and saved.state == AppInstanceState.TERMINATED:
		continue
		instance_registration: Optional[AppRegistration]
		if filter_registration is not None:
		instance_registration = filter_registration
		@@ -1010,6 +1015,47 @@ class EdgeApplicationManagementService:
		_camara_app_provider(instance_registration),
		)
		)
		# A refusal before fulfillment has no SRM instance. OEG owns the accepted
		# attempt and its terminal event state, so keep that failure visible too.
		if self._app_instance_repo is not None and self._operation_repo is not None:
		failed = await self._app_instance_repo.list_failed_by_provider(app_provider_id)
		seen = {row.appInstanceId for row in result}
		region_zones: set[UUID] \| None = None
		if failed and region is not None:
		zones = await self.srm_client.get_zones(
		region=region, status=None, x_correlator=x_correlator
		)
		region_zones = {UUID(zone.id) for zone in zones}
		for saved in failed:
		if saved.app_instance_id in seen:
		continue
		if app_instance_id is not None and saved.app_instance_id != app_instance_id:
		continue
		if app_id is not None and saved.app_registration_id != resolved_app_id:
		continue
		if region_zones is not None and saved.edge_cloud_zone_id not in region_zones:
		continue
		if self._app_registration_repo is None:
		raise RuntimeError("AppRegistrationRepository is not available")
		registration = await self._app_registration_repo.get_by_id(
		saved.app_registration_id
		)
		operation = await self._operation_repo.get_by_id(saved.operation_id)
		if registration is None or operation is None:
		continue
		name = operation.metadata.get("name") or operation.metadata.get(
		"app_deployment_name"
		)
		result.append(
		AppInstanceInfo(
		appInstanceId=saved.app_instance_id,
		name=str(name or saved.app_instance_id),
		appId=registration.app_id,
		appProvider=_camara_app_provider(registration, app_provider_id),
		status=AppInstanceStatus.FAILED,
		edgeCloudZoneId=saved.edge_cloud_zone_id,
		)
		)
		return result

		async def delete_app_instance(
		@@ -1575,6 +1621,21 @@ class EdgeApplicationManagementService:
		return None
		return app_registration, catalog

		async def _deleted_here(self, catalog: SRMCatalogPayload) -> bool:
		"""
		SRM only retires the specification of an app that ever had instances, and still
		lists it. The app was deleted once its registration here was, which is also why
		GET /apps/{appId} answers 404 for it.
		"""
		if self._app_registration_repo is None:
		return False
		try:
		registration_id = UUID(catalog.service_specification.id)
		except ValueError:
		return False
		registration = await self._app_registration_repo.get_by_id(registration_id)
		return registration is not None and registration.status == AppRegistrationStatus.DELETED

		async def _resolve_app_id(self, app_registration_id: UUID) -> Optional[UUID]:
		if self._app_registration_repo is None:
		return None

src/open_exposure_gateway/ports/database/instances.py

+4 −0

Changes for src/open_exposure_gateway/ports/database/instances.py: 4 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -11,6 +11,10 @@ class AppInstanceRepository(ABC):
		async def get_by_id(self, app_instance_id: UUID) -> AppInstance \| None:
		pass

		@abstractmethod
		async def list_failed_by_provider(self, app_provider_id: str) -> list[AppInstance]:
		pass

		@abstractmethod
		async def get_by_operation_id(self, operation_id: UUID) -> AppInstance \| None:
		pass

tests/integration/test_postgres.py

+27 −0

Changes for tests/integration/test_postgres.py: 27 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -731,3 +731,30 @@ async def test_qod_session_repo_updates_updated_at_on_second_save(db_engine: Asy
		assert updated.updated_at is not None
		assert updated.updated_at > updated.created_at
		assert updated.state == QodSessionState.AVAILABLE


		async def test_failed_instance_listing_scopes_by_operation_provider(
		db_session: AsyncSession,
		) -> None:
		registration = await SqlAppRegistrationRepository(db_session).save(_app_registration())
		operations = SqlOperationRepository(db_session)
		instances = SqlAppInstanceRepository(db_session)
		expected = []
		for provider, state in [
		("provider-a", AppInstanceState.FAILED),
		("provider-b", AppInstanceState.FAILED),
		("provider-a", AppInstanceState.TERMINATED),
		("provider-a", AppInstanceState.READY),
		]:
		operation = _operation(app_registration_id=registration.app_registration_id)
		operation.app_provider_id = provider
		await operations.save(operation)
		instance = _app_instance(operation.operation_id, registration.app_registration_id)
		instance.state = state
		await instances.save(instance)
		if provider == "provider-a" and state == AppInstanceState.FAILED:
		expected.append(instance.app_instance_id)
		assert [
		i.app_instance_id for i in await instances.list_failed_by_provider("provider-a")
		] == expected
		assert await instances.list_failed_by_provider("nobody") == []

tests/unit/fakes.py

+13 −1

Changes for tests/unit/fakes.py: 13 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -608,8 +608,20 @@ class FakeOperationRepository(OperationRepository):


		class FakeAppInstanceRepository(AppInstanceRepository):
		def __init__(self) -> None:
		def __init__(self, operations: FakeOperationRepository \| None = None) -> None:
		self.rows: dict[UUID, AppInstance] = {}
		self.operations = operations

		async def list_failed_by_provider(self, app_provider_id: str) -> list[AppInstance]:
		if self.operations is None:
		return []
		return [
		row.model_copy(deep=True)
		for row in self.rows.values()
		if row.state == AppInstanceState.FAILED
		and (operation := self.operations.rows.get(row.operation_id)) is not None
		and operation.app_provider_id == app_provider_id
		]

		async def get_by_id(self, app_instance_id: UUID) -> AppInstance \| None:
		found = self.rows.get(app_instance_id)