Merge branch 'fix/delete-failed-before-start' into 'develop' (e765f109) · Commits · OOP / code / open-exposure-gateway

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
		@@ -278,6 +278,7 @@ def _synthetic_failure_completion(event: SRMOperationStatus) -> SRMOperationComp


		_PREVIOUSLY_FAILED_APP_INSTANCE_IDS_KEY = "previously_failed_app_instance_ids"
		_FAILED_BEFORE_START_KEY = "failed_before_start"


		def _previously_failed_instance_ids(operation: Operation) -> frozenset[str]:
		@@ -861,6 +862,28 @@ class EdgeApplicationManagementService:
		if instance.state not in (AppInstanceState.TERMINATING, AppInstanceState.TERMINATED)
		]

		if deploy_operation.metadata.get(_FAILED_BEFORE_START_KEY):
		never_created = [
		instance for instance in live_instances if instance.state == AppInstanceState.FAILED
		]
		if never_created:
		for instance in never_created:
		logger.info(
		"never_created_app_instance_terminated_locally",
		app_instance_id=str(instance.app_instance_id),
		operation_id=str(instance.operation_id),
		app_deployment_id=str(app_deployment_id),
		)
		await self._app_instance_repo.save(
		instance.model_copy(update={"state": AppInstanceState.TERMINATED})
		)
		app_instances = await self._app_instance_repo.list_by_app_deployment_id(
		app_deployment_id
		)
		live_instances = [
		instance for instance in live_instances if instance not in never_created
		]

		if not live_instances:
		# Nothing for SRM to terminate, so no completion will arrive: settle here.
		operation_id, correlation_id, _ = self._new_operation_metadata(x_correlator)
		@@ -1077,6 +1100,12 @@ class EdgeApplicationManagementService:
		if deploy_operation is None or deploy_operation.app_provider_id != app_provider_id:
		raise NotFoundException(message=f"App instance {app_instance_id} not found")

		if app_instance.state == AppInstanceState.FAILED and deploy_operation.metadata.get(
		_FAILED_BEFORE_START_KEY
		):
		await self._terminate_never_created_instance(app_instance)
		return

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		command = build_terminate_instance_command(
		app_instance_id=app_instance_id,
		@@ -1113,6 +1142,36 @@ class EdgeApplicationManagementService:
		"Failed to publish app instance termination command",
		)

		async def _terminate_never_created_instance(self, app_instance: AppInstance) -> None:
		assert self._app_instance_repo is not None
		logger.info(
		"never_created_app_instance_terminated_locally",
		app_instance_id=str(app_instance.app_instance_id),
		operation_id=str(app_instance.operation_id),
		)
		saved = await self._app_instance_repo.save(
		app_instance.model_copy(update={"state": AppInstanceState.TERMINATED})
		)
		await self._deliver_callbacks(
		saved.operation_id, datetime.now(timezone.utc).isoformat(), [saved]
		)

		async def _mark_deploy_failed_before_start(self, operation_id: UUID) -> None:
		assert self._operation_repo is not None
		operation = await self._operation_repo.get_by_id(operation_id)
		if (
		operation is None
		or operation.operation_type != OperationType.DEPLOY
		or operation.status != OperationStatus.FAILED
		or operation.metadata.get(_FAILED_BEFORE_START_KEY)
		):
		return
		await self._operation_repo.save(
		operation.model_copy(
		update={"metadata": {**operation.metadata, _FAILED_BEFORE_START_KEY: True}}
		)
		)

		async def handle_completed(self, event: SRMOperationCompleted) -> None:
		if self._operation_repo is None:
		raise RuntimeError("OperationRepository is not available")
		@@ -1324,6 +1383,8 @@ class EdgeApplicationManagementService:
		state=event.state,
		)
		await self.handle_completed(_synthetic_failure_completion(event))
		if event.state == "failed_before_start":
		await self._mark_deploy_failed_before_start(UUID(event.operation_id))
		return

		operation_id = UUID(event.operation_id)

tests/unit/test_eam_service.py

+222 −0

Changes for tests/unit/test_eam_service.py: 222 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -2562,6 +2562,69 @@ class TestDeleteAppInstance:
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)

		async def _mark_failed_before_start(
		self,
		app_instance_repo: FakeAppInstanceRepository,
		operation_repo: FakeOperationRepository,
		) -> None:
		await self._mark_failed(app_instance_repo)
		instance = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert instance is not None
		deploy_operation = await operation_repo.get_by_id(instance.operation_id)
		assert deploy_operation is not None
		await operation_repo.save(
		deploy_operation.model_copy(
		update={
		"status": OperationStatus.FAILED,
		"metadata": {"failed_before_start": True},
		}
		)
		)

		async def test_instance_failed_before_start_is_terminated_without_srm(
		self,
		service: EdgeApplicationManagementService,
		publisher: AsyncMock,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._mark_failed_before_start(app_instance_repo, operation_repo)

		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)

		publisher.publish.assert_not_called()
		assert _terminate_operations(operation_repo) == []
		instance = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert instance is not None
		assert instance.state == AppInstanceState.TERMINATED
		with pytest.raises(NotFoundException):
		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)

		async def test_instance_failed_before_start_is_still_hidden_from_other_provider(
		self,
		service: EdgeApplicationManagementService,
		publisher: AsyncMock,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._mark_failed_before_start(app_instance_repo, operation_repo)

		with pytest.raises(NotFoundException):
		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID,
		tenant_id="tenant-1",
		app_provider_id="other-provider",
		)

		publisher.publish.assert_not_called()
		instance = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert instance is not None
		assert instance.state == AppInstanceState.FAILED

		async def test_ready_instance_with_failed_terminate_stays_failed(
		self,
		service: EdgeApplicationManagementService,
		@@ -2883,6 +2946,134 @@ class TestDeleteAppDeployment:
		assert deployment is not None
		assert deployment.state == AppDeploymentState.TERMINATED

		async def _mark_deploy_failed_before_start(
		self, operation_repo: FakeOperationRepository
		) -> None:
		deploy_operation = await operation_repo.get_by_id(self.DEPLOY_OPERATION_ID)
		assert deploy_operation is not None
		await operation_repo.save(
		deploy_operation.model_copy(
		update={
		"status": OperationStatus.FAILED,
		"metadata": {"failed_before_start": True},
		}
		)
		)

		async def test_deployment_failed_before_start_is_terminated_without_srm(
		self,
		service: EdgeApplicationManagementService,
		publisher: AsyncMock,
		operation_repo: FakeOperationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._seed_deployment(
		app_deployment_repo,
		app_instance_repo,
		deployment_state=AppDeploymentState.FAILED,
		instance_states={
		self.INSTANCE_A: AppInstanceState.FAILED,
		self.INSTANCE_B: AppInstanceState.FAILED,
		},
		)
		await self._mark_deploy_failed_before_start(operation_repo)

		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)

		publisher.publish.assert_not_called()
		(operation,) = _terminate_operations(operation_repo)
		assert operation.status == OperationStatus.COMPLETED
		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_failed_deployment_without_before_start_flag_still_goes_to_srm(
		self,
		service: EdgeApplicationManagementService,
		publisher: AsyncMock,
		operation_repo: FakeOperationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._seed_deployment(
		app_deployment_repo,
		app_instance_repo,
		deployment_state=AppDeploymentState.FAILED,
		instance_states={
		self.INSTANCE_A: AppInstanceState.FAILED,
		self.INSTANCE_B: AppInstanceState.FAILED,
		},
		)

		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)

		assert publisher.publish.call_count == 2
		(operation,) = _terminate_operations(operation_repo)
		assert operation.status == OperationStatus.PENDING

		async def test_deployment_failed_before_start_callback_reports_terminated_instances(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_registration_repo: FakeAppRegistrationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		callback_delivery_port: FakeCallbackDeliveryPort,
		) -> None:
		await app_registration_repo.save(
		AppRegistration(
		app_registration_id=APP_REGISTRATION_ID,
		app_id=APP_ID,
		tenant_id="tenant-1",
		name="myvideoapp",
		version="1.0.0",
		package_type=PackageType.HELM,
		status=AppRegistrationStatus.REGISTERED,
		)
		)
		await self._seed_deployment(
		app_deployment_repo,
		app_instance_repo,
		deployment_state=AppDeploymentState.FAILED,
		instance_states={
		self.INSTANCE_A: AppInstanceState.FAILED,
		self.INSTANCE_B: AppInstanceState.FAILED,
		},
		)
		await self._mark_deploy_failed_before_start(operation_repo)
		await callback_registration_repo.save(
		CallbackRegistration(
		id=uuid4(),
		operation_id=self.DEPLOY_OPERATION_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,
		)
		)

		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)

		[(_, events)] = callback_delivery_port.delivered
		assert isinstance(events, list)
		assert {e.data.appInstanceId for e in events} == {self.INSTANCE_A, self.INSTANCE_B}
		assert all(e.data.status == "unknown" for e in events)

		async def test_no_live_instances_still_fires_deployment_status_change_callback(
		self,
		service: EdgeApplicationManagementService,
		@@ -6192,6 +6383,37 @@ class TestHandleOperationStatus:
		assert updated_instance is not None
		assert updated_instance.state == AppInstanceState.FAILED

		@pytest.mark.parametrize(
		("failure_state", "flagged"), [("failed_before_start", True), ("failed", False)]
		)
		async def test_only_failed_before_start_flags_deploy_operation(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		failure_state: Literal["failed", "failed_before_start"],
		flagged: bool,
		) -> None:
		await self._seed_operation(operation_repo, OperationStatus.PENDING)

		await service.handle_operation_status(self._status_event(failure_state))

		updated = await operation_repo.get_by_id(self.OPERATION_ID)
		assert updated is not None
		assert updated.metadata.get("failed_before_start", False) is flagged

		async def test_failed_before_start_does_not_flag_terminate_operation(
		self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository
		) -> None:
		await self._seed_operation(
		operation_repo, OperationStatus.IN_PROGRESS, operation_type=OperationType.TERMINATE
		)

		await service.handle_operation_status(self._status_event("failed_before_start"))

		updated = await operation_repo.get_by_id(self.OPERATION_ID)
		assert updated is not None
		assert "failed_before_start" not in updated.metadata

		async def test_failed_moves_in_progress_terminate_operation_to_failed(
		self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository
		) -> None: