feat: finalise deployments on DELETE /deployments completion (ff208a17) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/edge_application_management_service.py

+47 −25

Original line number	Diff line number	Diff line
		@@ -107,6 +107,15 @@ _APP_DEPLOYMENT_COMPLETION_STATE_MAP: dict[OperationStatus, AppDeploymentState]
		OperationStatus.FAILED: AppDeploymentState.FAILED,
		}

		# Same rollup, but for a DELETE /deployments teardown operation: a clean
		# completion means the aggregate reached TERMINATED, not READY. partial/failed
		# keep their meaning (some / every zone failed to tear down).
		_APP_DEPLOYMENT_TERMINATE_STATE_MAP: dict[OperationStatus, AppDeploymentState] = {
		OperationStatus.COMPLETED: AppDeploymentState.TERMINATED,
		OperationStatus.PARTIALLY_COMPLETED: AppDeploymentState.PARTIAL,
		OperationStatus.FAILED: AppDeploymentState.FAILED,
		}

		_TERMINAL_OPERATION_STATUSES = frozenset(
		{
		OperationStatus.COMPLETED,
		@@ -927,21 +936,25 @@ class EdgeApplicationManagementService:
		elif status == OperationStatus.FAILED:
		# Total failure carries no instances[] entries to match against.
		if is_terminate:
		# Termination always targets exactly one instance (DELETE
		# /deployments doesn't exist yet, so there's no multi-instance
		# terminate operation to fan out over).
		raw_app_instance_id = operation.metadata.get("app_instance_id")
		app_instance = (
		await self._app_instance_repo.get_by_id(UUID(raw_app_instance_id))
		if raw_app_instance_id is not None
		else None
		)
		# DELETE /appinstances stashes one app_instance_id; DELETE
		# /deployments stashes the whole app_instance_ids list under one
		# aggregate operation. Fan out over whichever the operation
		# recorded so every targeted instance flips to failed.
		raw_ids = operation.metadata.get("app_instance_ids")
		if raw_ids is None:
		single = operation.metadata.get("app_instance_id")
		raw_ids = [single] if single is not None else []
		failed_instances: list[AppInstance] = []
		for raw_id in raw_ids:
		app_instance = await self._app_instance_repo.get_by_id(UUID(raw_id))
		if app_instance is None:
		logger.warning(
		"terminate_total_failure_missing_app_instance_id",
		operation_id=event.operation_id,
		app_instance_id=raw_id,
		)
		failed_instances = [app_instance] if app_instance is not None else []
		continue
		failed_instances.append(app_instance)
		else:
		# A deploy operation_id may own N pre-created rows (one per
		# zone for a multi-zone /deployments request); all of them
		@@ -957,20 +970,29 @@ 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, but the
		# Optional cache rollup for the app_deployments aggregate row
		# (persistence-model.md). A deploy operation owns the row directly
		# (get_by_operation_id); a DELETE /deployments teardown is its own
		# operation, so it carries the deployment id in metadata instead. A
		# single-zone /appinstances operation resolves neither -- soft no-op,
		# same optional-dependency pattern as the callback repos below. The
		# _is_final guard mirrors the instances above so a stale redelivery
		# can't revive a terminated deployment once DELETE lands.
		# can't revive a terminated deployment.
		if self._app_deployment_repo is not None:
		if is_terminate:
		raw_deployment_id = operation.metadata.get("app_deployment_id")
		app_deployment = (
		await self._app_deployment_repo.get_by_id(UUID(raw_deployment_id))
		if raw_deployment_id is not None
		else None
		)
		deployment_state_map = _APP_DEPLOYMENT_TERMINATE_STATE_MAP
		else:
		app_deployment = await self._app_deployment_repo.get_by_operation_id(operation_id)
		deployment_state_map = _APP_DEPLOYMENT_COMPLETION_STATE_MAP
		if app_deployment is not None and not _deployment_is_final(app_deployment):
		await self._app_deployment_repo.save(
		app_deployment.model_copy(
		update={"state": _APP_DEPLOYMENT_COMPLETION_STATE_MAP[status]}
		)
		app_deployment.model_copy(update={"state": deployment_state_map[status]})
		)

		if updated_instances:

tests/unit/test_eam_service.py

+335 −0

Original line number	Diff line number	Diff line
		@@ -3706,3 +3706,338 @@ class TestHandleCompleted:
		rows = list(callback_delivery_repo.rows.values())
		assert len(rows) == 1
		assert rows[0].state == "delivered"


		class TestDeploymentTerminateCompletion:
		"""handle_completed for a DELETE /deployments teardown operation.

		The teardown is its own operations row (OperationType.TERMINATE) carrying
		app_deployment_id + app_instance_ids in metadata; the app_deployments and
		app_instances rows still point at the original DEPLOY operation.
		"""

		DEPLOY_OP_ID = UUID("dddddddd-dddd-dddd-dddd-dddddddddd01")
		TERMINATE_OP_ID = UUID("dddddddd-dddd-dddd-dddd-dddddddddd02")
		INSTANCE_A = UUID("cccccccc-0000-4000-8000-0000000000a1")
		INSTANCE_B = UUID("cccccccc-0000-4000-8000-0000000000b2")

		async def _seed(
		self,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		instance_ids: list[UUID],
		*,
		terminate_status: OperationStatus = OperationStatus.PENDING,
		app_registration_repo: FakeAppRegistrationRepository \| None = None,
		) -> UUID:
		app_registration_id = uuid4()
		app_id = uuid4()
		if app_registration_repo is not None:
		await app_registration_repo.save(
		AppRegistration(
		app_registration_id=app_registration_id,
		app_id=app_id,
		tenant_id="tenant-1",
		name="video_analytics_eu",
		version="1.0.0",
		package_type=PackageType.HELM,
		status=AppRegistrationStatus.REGISTERED,
		)
		)
		await operation_repo.save(
		Operation(
		operation_id=self.TERMINATE_OP_ID,
		correlation_id="corr-1",
		tenant_id="tenant-1",
		app_provider_id="provider-1",
		operation_type=OperationType.TERMINATE,
		status=terminate_status,
		subject=Subject.TASK_TERMINATE,
		app_registration_id=app_registration_id,
		metadata={
		"app_deployment_id": str(DEPLOYMENT_ID),
		"app_instance_ids": [str(i) for i in instance_ids],
		},
		)
		)
		await app_deployment_repo.save(
		AppDeployment(
		app_deployment_id=DEPLOYMENT_ID,
		operation_id=self.DEPLOY_OP_ID,
		app_registration_id=app_registration_id,
		app_deployment_name="video_analytics_eu",
		edge_cloud_zones=[ZONE_ID],
		state=AppDeploymentState.TERMINATING,
		)
		)
		for instance_id in instance_ids:
		await app_instance_repo.save(
		AppInstance(
		app_instance_id=instance_id,
		operation_id=self.DEPLOY_OP_ID,
		app_registration_id=app_registration_id,
		edge_cloud_zone_id=ZONE_ID,
		state=AppInstanceState.TERMINATING,
		app_deployment_id=DEPLOYMENT_ID,
		)
		)
		return app_id

		def _completed_event(
		self,
		status: str,
		instances: list[SRMCompletedInstance],
		error: dict[str, Any] \| None = None,
		) -> SRMOperationCompleted:
		return SRMOperationCompleted(
		schema_version="1.0",
		operation_id=str(self.TERMINATE_OP_ID),
		status=status, # type: ignore[arg-type]
		instances=instances,
		error=error,
		correlation_id="corr-1",
		completed_at="2026-07-04T10:02:35+00:00",
		)

		async def test_full_completion_rolls_deployment_and_instances_to_terminated(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await self._seed(
		operation_repo,
		app_instance_repo,
		app_deployment_repo,
		[self.INSTANCE_A, self.INSTANCE_B],
		)
		event = self._completed_event(
		"completed",
		[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_A),
		zone_id=str(ZONE_ID),
		status="completed",
		),
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_B),
		zone_id=str(ZONE_ID),
		status="completed",
		),
		],
		)

		await service.handle_completed(event)

		deployment = await app_deployment_repo.get_by_id(DEPLOYMENT_ID)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.TERMINATED
		for instance_id in (self.INSTANCE_A, self.INSTANCE_B):
		instance = await app_instance_repo.get_by_id(instance_id)
		assert instance is not None
		assert instance.state == AppInstanceState.TERMINATED

		async def test_partial_completion_rolls_deployment_to_partial(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await self._seed(
		operation_repo,
		app_instance_repo,
		app_deployment_repo,
		[self.INSTANCE_A, self.INSTANCE_B],
		)
		event = self._completed_event(
		"partially_completed",
		[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_A),
		zone_id=str(ZONE_ID),
		status="completed",
		),
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_B),
		zone_id=str(ZONE_ID),
		status="failed",
		error={"title": "Termination Failed", "status": 503},
		),
		],
		)

		await service.handle_completed(event)

		deployment = await app_deployment_repo.get_by_id(DEPLOYMENT_ID)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.PARTIAL
		a = await app_instance_repo.get_by_id(self.INSTANCE_A)
		b = await app_instance_repo.get_by_id(self.INSTANCE_B)
		assert a is not None and a.state == AppInstanceState.TERMINATED
		assert b is not None and b.state == AppInstanceState.FAILED

		async def test_total_failure_fans_out_over_every_stashed_instance(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await self._seed(
		operation_repo,
		app_instance_repo,
		app_deployment_repo,
		[self.INSTANCE_A, self.INSTANCE_B],
		)
		event = self._completed_event(
		"failed",
		[],
		error={"title": "Termination Failed", "status": 503, "detail": "backend down"},
		)

		await service.handle_completed(event)

		deployment = await app_deployment_repo.get_by_id(DEPLOYMENT_ID)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.FAILED
		for instance_id in (self.INSTANCE_A, self.INSTANCE_B):
		instance = await app_instance_repo.get_by_id(instance_id)
		assert instance is not None
		assert instance.state == AppInstanceState.FAILED

		async def test_redelivered_completion_does_not_disturb_terminated_deployment(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await self._seed(
		operation_repo,
		app_instance_repo,
		app_deployment_repo,
		[self.INSTANCE_A],
		terminate_status=OperationStatus.COMPLETED,
		)
		await app_deployment_repo.save(
		(await app_deployment_repo.get_by_id(DEPLOYMENT_ID)).model_copy( # type: ignore[union-attr]
		update={"state": AppDeploymentState.TERMINATED}
		)
		)
		event = self._completed_event(
		"completed",
		[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_A),
		zone_id=str(ZONE_ID),
		status="completed",
		)
		],
		)

		await service.handle_completed(event)

		deployment = await app_deployment_repo.get_by_id(DEPLOYMENT_ID)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.TERMINATED

		async def test_delivers_one_deployment_status_change_array(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_registration_repo: FakeAppRegistrationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		callback_delivery_port: FakeCallbackDeliveryPort,
		callback_delivery_repo: FakeCallbackDeliveryRepository,
		) -> None:
		await self._seed(
		operation_repo,
		app_instance_repo,
		app_deployment_repo,
		[self.INSTANCE_A, self.INSTANCE_B],
		app_registration_repo=app_registration_repo,
		)
		# The subscription was registered against the original DEPLOY operation.
		await callback_registration_repo.save(
		CallbackRegistration(
		id=uuid4(),
		operation_id=self.DEPLOY_OP_ID,
		tenant_id="tenant-1",
		api_family="edge-application-management",
		sink="https://client.example.com/callback",
		event_types=[
		"org.camaraproject.edge-application-management.v0.app-deployment-status-change"
		],
		is_active=True,
		)
		)
		event = self._completed_event(
		"completed",
		[
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_A),
		zone_id=str(ZONE_ID),
		status="completed",
		),
		SRMCompletedInstance(
		service_instance_id=str(self.INSTANCE_B),
		zone_id=str(ZONE_ID),
		status="completed",
		),
		],
		)

		await service.handle_completed(event)

		assert len(callback_delivery_port.delivered) == 1
		_, payload = callback_delivery_port.delivered[0]
		assert isinstance(payload, list)
		assert {ce.data.appDeploymentId for ce in payload} == {DEPLOYMENT_ID}
		assert len(list(callback_delivery_repo.rows.values())) == 1

		async def test_single_instance_delete_total_failure_still_marks_its_instance(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		"""Regression: DELETE /appinstances stashes the legacy single
		app_instance_id key, not app_instance_ids -- the fan-out must still
		resolve it."""
		await operation_repo.save(
		Operation(
		operation_id=self.TERMINATE_OP_ID,
		correlation_id="corr-1",
		tenant_id="tenant-1",
		app_provider_id="provider-1",
		operation_type=OperationType.TERMINATE,
		status=OperationStatus.PENDING,
		subject=Subject.TASK_TERMINATE,
		metadata={"app_instance_id": str(self.INSTANCE_A)},
		)
		)
		await app_instance_repo.save(
		AppInstance(
		app_instance_id=self.INSTANCE_A,
		operation_id=uuid4(),
		app_registration_id=uuid4(),
		edge_cloud_zone_id=ZONE_ID,
		state=AppInstanceState.TERMINATING,
		)
		)
		event = self._completed_event(
		"failed",
		[],
		error={"title": "Termination Failed", "status": 503, "detail": "backend down"},
		)

		await service.handle_completed(event)

		instance = await app_instance_repo.get_by_id(self.INSTANCE_A)
		assert instance is not None
		assert instance.state == AppInstanceState.FAILED