feat: roll up app_deployments.state on operation completion (6a2c2017) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/edge_application_management_service.py

+27 −0

Original line number	Diff line number	Diff line
		@@ -90,6 +90,16 @@ _APP_INSTANCE_COMPLETION_STATE_MAP: dict[str, AppInstanceState] = {
		"failed": AppInstanceState.FAILED,
		}

		# Cached rollup for app_deployments.state (persistence-model.md); only
		# multi-zone POST /deployments operations have a matching row -- a single-zone
		# /appinstances operation_id simply won't resolve one, so this map only ever
		# gets consulted when there is one.
		_APP_DEPLOYMENT_COMPLETION_STATE_MAP: dict[OperationStatus, AppDeploymentState] = {
		OperationStatus.COMPLETED: AppDeploymentState.READY,
		OperationStatus.PARTIALLY_COMPLETED: AppDeploymentState.PARTIAL,
		OperationStatus.FAILED: AppDeploymentState.FAILED,
		}

		_TERMINAL_OPERATION_STATUSES = frozenset(
		{
		OperationStatus.COMPLETED,
		@@ -752,6 +762,23 @@ class EdgeApplicationManagementService:
		)
		updated_instances.append(saved)

		# Optional cache rollup, multi-zone deployments only (persistence-model.md).
		# A single-zone /appinstances operation_id has no matching row, so this
		# is a soft no-op for the existing path -- same optional-dependency
		# pattern as the callback repos below. DELETE /deployments doesn't
		# exist yet, so app_deployments.state can only be INSTANTIATING here;
		# once it does, this will need the same _is_final-style guard the
		# instances above already have, to avoid reviving a terminated
		# deployment on a stale redelivery.
		if self._app_deployment_repo is not None:
		app_deployment = await self._app_deployment_repo.get_by_operation_id(operation_id)
		if app_deployment is not None:
		await self._app_deployment_repo.save(
		app_deployment.model_copy(
		update={"state": _APP_DEPLOYMENT_COMPLETION_STATE_MAP[status]}
		)
		)

		if updated_instances:
		await self._deliver_callbacks(operation_id, event.completed_at, updated_instances)

tests/unit/test_eam_service.py

+176 −0

Original line number	Diff line number	Diff line
		@@ -50,6 +50,7 @@ from open_exposure_gateway.domain.edge_application_management import (
		Subject,
		)
		from open_exposure_gateway.domain.models import (
		AppDeployment,
		AppDeploymentState,
		AppInstance,
		AppInstanceState,
		@@ -1700,6 +1701,20 @@ class TestHandleCompleted:
		)
		)

		async def _seed_instantiating_app_deployment(
		self, app_deployment_repo: FakeAppDeploymentRepository, app_deployment_id: UUID
		) -> None:
		await app_deployment_repo.save(
		AppDeployment(
		app_deployment_id=app_deployment_id,
		operation_id=self.OPERATION_ID,
		app_registration_id=uuid4(),
		app_deployment_name="video_analytics_eu",
		edge_cloud_zones=[ZONE_ID],
		state=AppDeploymentState.INSTANTIATING,
		)
		)

		async def _seed_app_instance_with_registration(
		self,
		app_registration_repo: FakeAppRegistrationRepository,
		@@ -2087,6 +2102,167 @@ class TestHandleCompleted:
		assert unchanged is not None
		assert unchanged.state == AppInstanceState.TERMINATED

		async def test_rolls_up_app_deployment_to_ready_on_full_completion(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		deployment_id = uuid4()
		await self._seed_pending_operation(operation_repo)
		await self._seed_instantiating_app_instance(app_instance_repo, INSTANCE_ID)
		await self._seed_instantiating_app_deployment(app_deployment_repo, deployment_id)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(INSTANCE_ID), zone_id=str(ZONE_ID), status="completed"
		)
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

		updated = await app_deployment_repo.get_by_id(deployment_id)
		assert updated is not None
		assert updated.state == AppDeploymentState.READY

		async def test_rolls_up_app_deployment_to_partial_on_partial_completion(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		ready_id = UUID("9a3f1c22-0000-4000-8000-00000000000a")
		failed_id = UUID("9a3f1c22-0000-4000-8000-00000000000b")
		deployment_id = uuid4()
		await self._seed_pending_operation(operation_repo)
		await self._seed_instantiating_app_instance(app_instance_repo, ready_id)
		await self._seed_instantiating_app_instance(app_instance_repo, failed_id)
		await self._seed_instantiating_app_deployment(app_deployment_repo, deployment_id)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="partially_completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(ready_id), zone_id=str(ZONE_ID), status="completed"
		),
		SRMCompletedInstance(
		service_instance_id=str(failed_id),
		zone_id=str(ZONE_ID),
		status="failed",
		error={"title": "Zone Capacity Exceeded", "status": 503},
		),
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

		updated = await app_deployment_repo.get_by_id(deployment_id)
		assert updated is not None
		assert updated.state == AppDeploymentState.PARTIAL

		async def test_rolls_up_app_deployment_to_failed_on_total_failure(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		deployment_id = uuid4()
		await self._seed_pending_operation(operation_repo)
		await self._seed_instantiating_app_instance(app_instance_repo, INSTANCE_ID)
		await self._seed_instantiating_app_deployment(app_deployment_repo, deployment_id)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="failed",
		instances=[],
		error={"title": "Zone Capacity Exceeded", "status": 503},
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

		updated = await app_deployment_repo.get_by_id(deployment_id)
		assert updated is not None
		assert updated.state == AppDeploymentState.FAILED

		async def test_single_zone_operation_leaves_app_deployments_untouched(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		"""A single-zone /appinstances operation_id has no matching
		app_deployments row -- the rollup must be a silent no-op, not an
		error, and must not fabricate a row."""
		await self._seed_pending_operation(operation_repo)
		await self._seed_instantiating_app_instance(app_instance_repo, INSTANCE_ID)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(INSTANCE_ID), zone_id=str(ZONE_ID), status="completed"
		)
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

		assert app_deployment_repo.rows == {}

		async def test_completion_succeeds_when_app_deployment_repo_unavailable(
		self,
		srm_client: AsyncMock,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		"""app_deployment_repo is optional -- a service built without it (e.g.
		an older deployment of this component) must keep completing
		single-zone instances rather than raising."""
		service = EdgeApplicationManagementService(
		srm_client=srm_client,
		operation_repo=operation_repo,
		app_instance_repo=app_instance_repo,
		app_deployment_repo=None,
		)
		await self._seed_pending_operation(operation_repo)
		await self._seed_instantiating_app_instance(app_instance_repo, INSTANCE_ID)
		event = SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.OPERATION_ID),
		status="completed",
		instances=[
		SRMCompletedInstance(
		service_instance_id=str(INSTANCE_ID), zone_id=str(ZONE_ID), status="completed"
		)
		],
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		await service.handle_completed(event)

		updated = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert updated is not None
		assert updated.state == AppInstanceState.READY

		async def test_failure_expressed_only_via_instances_is_processed(
		self,
		service: EdgeApplicationManagementService,