feat: implement exists_in_zone method in AppInstanceRepository and... (46e3f87d) · Commits · OOP / code / open-exposure-gateway

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

+9 −0

Original line number	Diff line number	Diff line
		@@ -33,6 +33,15 @@ class SqlAppInstanceRepository(AppInstanceRepository):
		row = await self._session.scalar(stmt.limit(1))
		return row is not None

		async def exists_in_zone(self, app_registration_id: UUID, edge_cloud_zone_id: UUID) -> bool:
		stmt = select(AppInstanceRow.app_instance_id).where(
		AppInstanceRow.app_registration_id == app_registration_id,
		AppInstanceRow.edge_cloud_zone_id == edge_cloud_zone_id,
		AppInstanceRow.state.notin_(_TERMINAL_STATES),
		)
		row = await self._session.scalar(stmt.limit(1))
		return row is not None

		async def save(self, app_instance: AppInstance) -> AppInstance:
		merged = await self._session.merge(AppInstanceMapper.to_row(app_instance))
		await self._session.flush()

src/open_exposure_gateway/application/services/edge_application_management_service.py

+12 −0

Original line number	Diff line number	Diff line
		@@ -331,6 +331,13 @@ class EdgeApplicationManagementService:
		if app_registration is None:
		raise BadRequestException(message=f"App {request.appId} is not registered")

		if await self._app_instance_repo.exists_in_zone(
		app_registration.app_registration_id, request.edgeCloudZoneId
		):
		raise AlreadyExistsException(
		message="Application already instantiated in the given Edge Cloud Zone"
		)

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		app_instance_id = uuid4()
		translation = build_app_deployment_translation(
		@@ -375,6 +382,11 @@ class EdgeApplicationManagementService:
		if request.subscriptionRequest is not None:
		if self._callback_registration_repo is None:
		raise RuntimeError("CallbackRegistrationRepository is not available")
		# TODO: the raw subscription.sinkCredential token is dropped here — the
		# secret:// ref below points at a store that does not exist yet, and
		# HttpCallbackClient sends no Authorization header, so any sink requiring
		# auth rejects every callback (ADR-0008 requires a sinkCredential bearer
		# token). Either persist the credential or reject sinkCredential with 501.
		subscription = request.subscriptionRequest
		await self._callback_registration_repo.save(
		CallbackRegistration(

src/open_exposure_gateway/ports/database/instances.py

+4 −0

Original line number	Diff line number	Diff line
		@@ -19,6 +19,10 @@ class AppInstanceRepository(ABC):
		async def exists_for_app_registration(self, app_registration_id: UUID) -> bool:
		pass

		@abstractmethod
		async def exists_in_zone(self, app_registration_id: UUID, edge_cloud_zone_id: UUID) -> bool:
		pass

		@abstractmethod
		async def save(self, app_instance: AppInstance) -> AppInstance:
		pass

tests/integration/test_postgres.py

+45 −0

Original line number	Diff line number	Diff line
		@@ -217,6 +217,51 @@ async def test_app_instance_repo_exists_for_app_registration_ignores_terminal_st
		assert await repo.exists_for_app_registration(registration.app_registration_id) is False


		async def test_app_instance_repo_exists_in_zone_matches_only_same_zone(
		db_session: AsyncSession,
		) -> None:
		registration = await SqlAppRegistrationRepository(db_session).save(_app_registration())
		operation = await SqlOperationRepository(db_session).save(
		_operation(app_registration_id=registration.app_registration_id)
		)
		repo = SqlAppInstanceRepository(db_session)
		instance = _app_instance(
		operation_id=operation.operation_id,
		app_registration_id=registration.app_registration_id,
		)

		await repo.save(instance)

		assert (
		await repo.exists_in_zone(registration.app_registration_id, instance.edge_cloud_zone_id)
		is True
		)
		assert await repo.exists_in_zone(registration.app_registration_id, uuid4()) is False


		@pytest.mark.parametrize("state", [AppInstanceState.FAILED, AppInstanceState.TERMINATED])
		async def test_app_instance_repo_exists_in_zone_ignores_terminal_states(
		db_session: AsyncSession, state: AppInstanceState
		) -> None:
		registration = await SqlAppRegistrationRepository(db_session).save(_app_registration())
		operation = await SqlOperationRepository(db_session).save(
		_operation(app_registration_id=registration.app_registration_id)
		)
		repo = SqlAppInstanceRepository(db_session)
		instance = _app_instance(
		operation_id=operation.operation_id,
		app_registration_id=registration.app_registration_id,
		)
		instance.state = state

		await repo.save(instance)

		assert (
		await repo.exists_in_zone(registration.app_registration_id, instance.edge_cloud_zone_id)
		is False
		)


		async def test_app_registration_delete_keeps_row_referenced_by_terminated_instance(
		db_session: AsyncSession,
		) -> None:

tests/unit/fakes.py

+9 −0

Original line number	Diff line number	Diff line
		@@ -367,6 +367,15 @@ class FakeAppInstanceRepository(AppInstanceRepository):
		for row in self.rows.values()
		)

		async def exists_in_zone(self, app_registration_id: UUID, edge_cloud_zone_id: UUID) -> bool:
		terminal_states = (AppInstanceState.FAILED, AppInstanceState.TERMINATED)
		return any(
		row.app_registration_id == app_registration_id
		and row.edge_cloud_zone_id == edge_cloud_zone_id
		and row.state not in terminal_states
		for row in self.rows.values()
		)

		async def save(self, app_instance: AppInstance) -> AppInstance:
		stored = app_instance.model_copy(deep=True)
		self.rows[stored.app_instance_id] = stored

Admin message