feat: implement DELETE /deployments request path (93b31ddd) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/api/camara/edge_application_management/vwip/router.py

+13 −3

Original line number	Diff line number	Diff line
		@@ -349,18 +349,28 @@ async def get_app_deployments(
		@router.delete(
		"/deployments/{appDeploymentId}",
		tags=["Application"],
		status_code=202,
		summary="Terminate an Application Deployment",
		responses=_responses(
		BadRequestException(),
		UnauthorizedException(),
		ForbiddenException(),
		NotFoundException(),
		NotImplementedException(),
		DownstreamServiceException(),
		),
		)
		async def delete_app_deployment(appDeploymentId: UUID) -> Any:
		raise NotImplementedException(message="Multi-zone deployment is not implemented")
		async def delete_app_deployment(
		appDeploymentId: UUID,
		service: EdgeAppService,
		caller: Caller,
		) -> Response:
		await service.delete_app_deployment(
		app_deployment_id=appDeploymentId,
		tenant_id=caller.tenant_id,
		app_provider_id=caller.app_provider_id,
		x_correlator=caller.x_correlator,
		)
		return Response(status_code=status.HTTP_202_ACCEPTED)


		@router.patch(

src/open_exposure_gateway/application/services/edge_application_management_service.py

+88 −0

Original line number	Diff line number	Diff line
		@@ -656,6 +656,94 @@ class EdgeApplicationManagementService:
		)
		return result

		async def delete_app_deployment(
		self,
		app_deployment_id: UUID,
		tenant_id: str,
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		) -> None:
		if self._operation_repo is None:
		raise RuntimeError("OperationRepository is not available")
		if self._app_instance_repo is None:
		raise RuntimeError("AppInstanceRepository is not available")
		if self._app_deployment_repo is None:
		raise RuntimeError("AppDeploymentRepository is not available")

		app_deployment = await self._app_deployment_repo.get_by_id(app_deployment_id)
		if app_deployment is None or app_deployment.state in (
		AppDeploymentState.TERMINATING,
		AppDeploymentState.TERMINATED,
		):
		raise NotFoundException(message=f"App deployment {app_deployment_id} not found")

		app_instances = await self._app_instance_repo.list_by_app_deployment_id(app_deployment_id)
		# Instances already gone (individually DELETEd via /appinstances, or a
		# prior failure) get no fresh terminate command -- SRM has nothing left
		# to tear down for them.
		live_instances = [
		instance
		for instance in app_instances
		if instance.state not in (AppInstanceState.TERMINATING, AppInstanceState.TERMINATED)
		]

		if not live_instances:
		# Nothing to hand to SRM, so no completion will ever arrive to
		# finalise the aggregate -- settle it here rather than leave the row
		# stuck at terminating.
		await self._app_deployment_repo.save(
		app_deployment.model_copy(update={"state": AppDeploymentState.TERMINATED})
		)
		return

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)

		# One aggregate operation for the whole deployment (mirrors QoD's
		# deactivate: teardown is its own operations row that carries the record
		# id in metadata). SRM fans N terminate commands sharing this id back
		# into a single event.srm.operation.completed. app_instance_ids is the
		# fallback handle_completed uses when a total failure carries no
		# instances[] to match against.
		await self._operation_repo.save(
		Operation(
		operation_id=operation_id,
		correlation_id=correlation_id,
		tenant_id=tenant_id,
		app_provider_id=app_provider_id,
		operation_type=OperationType.TERMINATE,
		status=OperationStatus.PENDING,
		subject=Subject.TASK_TERMINATE,
		app_registration_id=app_deployment.app_registration_id,
		metadata={
		"app_deployment_id": str(app_deployment_id),
		"app_instance_ids": [
		str(instance.app_instance_id) for instance in live_instances
		],
		},
		)
		)
		await self._app_deployment_repo.save(
		app_deployment.model_copy(update={"state": AppDeploymentState.TERMINATING})
		)
		for instance in live_instances:
		await self._app_instance_repo.save(
		instance.model_copy(update={"state": AppInstanceState.TERMINATING})
		)

		for instance in live_instances:
		command = build_terminate_instance_command(
		app_instance_id=instance.app_instance_id,
		operation_id=operation_id,
		app_provider_id=app_provider_id,
		correlation_id=correlation_id,
		requested_at=requested_at,
		)
		await self._publish(
		Subject.TASK_TERMINATE,
		command,
		"Failed to publish app deployment termination command",
		)

		async def get_app_instances(
		self,
		app_id: Optional[UUID] = None,

tests/unit/test_eam_endpoints.py

+31 −0

Original line number	Diff line number	Diff line
		@@ -21,6 +21,7 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas i
		from open_exposure_gateway.application.services.edge_application_management_service import (
		EdgeApplicationManagementService,
		)
		from open_exposure_gateway.core.exceptions import NotFoundException
		from open_exposure_gateway.dependencies import get_edge_app_service
		from open_exposure_gateway.main import app

		@@ -271,3 +272,33 @@ class TestGetAppDeployments:
		app_deployment_id=_DEPLOYMENT_ID,
		x_correlator=None,
		)


		class TestDeleteAppDeployment:
		def test_returns_202(self, client: TestClient) -> None:
		response = client.delete(f"{EAM_BASE}/deployments/{_DEPLOYMENT_ID}")
		assert response.status_code == 202

		def test_service_called_with_deployment_id_and_caller(
		self, client: TestClient, mock_eam_service: AsyncMock
		) -> None:
		client.delete(f"{EAM_BASE}/deployments/{_DEPLOYMENT_ID}")
		mock_eam_service.delete_app_deployment.assert_called_once_with(
		app_deployment_id=_DEPLOYMENT_ID,
		tenant_id=ANY,
		app_provider_id=ANY,
		x_correlator=ANY,
		)

		def test_unknown_deployment_returns_404(
		self, client: TestClient, mock_eam_service: AsyncMock
		) -> None:
		mock_eam_service.delete_app_deployment.side_effect = NotFoundException(
		message="App deployment not found"
		)
		response = client.delete(f"{EAM_BASE}/deployments/{_DEPLOYMENT_ID}")
		assert response.status_code == 404

		def test_malformed_deployment_id_returns_400(self, client: TestClient) -> None:
		response = client.delete(f"{EAM_BASE}/deployments/not-a-uuid")
		assert response.status_code == 400

tests/unit/test_eam_service.py

+252 −0

Original line number	Diff line number	Diff line
		@@ -1857,6 +1857,258 @@ class TestDeleteAppInstance:
		)


		class TestDeleteAppDeployment:
		ZONE_A = UUID("aaaaaaaa-1111-4000-8000-000000000001")
		ZONE_B = UUID("aaaaaaaa-2222-4000-8000-000000000002")
		INSTANCE_A = UUID("cccccccc-1111-4000-8000-000000000001")
		INSTANCE_B = UUID("cccccccc-2222-4000-8000-000000000002")
		DEPLOY_OPERATION_ID = UUID("22222222-2222-4000-8000-000000000002")

		async def _seed_deployment(
		self,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		deployment_state: AppDeploymentState = AppDeploymentState.READY,
		instance_states: dict[UUID, AppInstanceState] \| None = None,
		) -> None:
		instance_states = instance_states or {
		self.INSTANCE_A: AppInstanceState.READY,
		self.INSTANCE_B: AppInstanceState.READY,
		}
		await app_deployment_repo.save(
		AppDeployment(
		app_deployment_id=DEPLOYMENT_ID,
		operation_id=self.DEPLOY_OPERATION_ID,
		app_registration_id=APP_REGISTRATION_ID,
		app_deployment_name="video_analytics_eu",
		edge_cloud_zones=[self.ZONE_A, self.ZONE_B],
		state=deployment_state,
		)
		)
		zone_by_instance = {self.INSTANCE_A: self.ZONE_A, self.INSTANCE_B: self.ZONE_B}
		for instance_id, state in instance_states.items():
		await app_instance_repo.save(
		AppInstance(
		app_instance_id=instance_id,
		operation_id=self.DEPLOY_OPERATION_ID,
		app_registration_id=APP_REGISTRATION_ID,
		edge_cloud_zone_id=zone_by_instance[instance_id],
		state=state,
		app_deployment_id=DEPLOYMENT_ID,
		)
		)

		async def test_publishes_one_terminate_command_per_instance_sharing_one_operation(
		self,
		service: EdgeApplicationManagementService,
		publisher: AsyncMock,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._seed_deployment(app_deployment_repo, app_instance_repo)
		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID,
		tenant_id="tenant-1",
		app_provider_id="VideoAppsCo",
		x_correlator="corr-1",
		)
		assert publisher.publish.call_count == 2
		subjects = {call.args[0] for call in publisher.publish.call_args_list}
		payloads = [call.args[1] for call in publisher.publish.call_args_list]
		assert subjects == {Subject.TASK_TERMINATE}
		assert {p["service_instance_id"] for p in payloads} == {
		str(self.INSTANCE_A),
		str(self.INSTANCE_B),
		}
		assert len({p["operation_id"] for p in payloads}) == 1
		assert {p["correlation_id"] for p in payloads} == {"corr-1"}

		async def test_persists_one_pending_terminate_operation_with_ids_in_metadata(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._seed_deployment(app_deployment_repo, app_instance_repo)
		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)
		(operation,) = list(operation_repo.rows.values())
		assert operation.status == OperationStatus.PENDING
		assert operation.operation_type == OperationType.TERMINATE
		assert operation.subject == Subject.TASK_TERMINATE
		assert operation.app_registration_id == APP_REGISTRATION_ID
		assert operation.metadata["app_deployment_id"] == str(DEPLOYMENT_ID)
		assert set(operation.metadata["app_instance_ids"]) == {
		str(self.INSTANCE_A),
		str(self.INSTANCE_B),
		}

		async def test_moves_deployment_and_every_instance_to_terminating(
		self,
		service: EdgeApplicationManagementService,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._seed_deployment(app_deployment_repo, app_instance_repo)
		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="p"
		)
		deployment = await app_deployment_repo.get_by_id(DEPLOYMENT_ID)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.TERMINATING
		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.TERMINATING

		async def test_raises_not_found_for_unknown_deployment(
		self, service: EdgeApplicationManagementService
		) -> None:
		with pytest.raises(NotFoundException):
		await service.delete_app_deployment(
		app_deployment_id=uuid4(), tenant_id="tenant-1", app_provider_id="p"
		)

		@pytest.mark.parametrize(
		"state", [AppDeploymentState.TERMINATING, AppDeploymentState.TERMINATED]
		)
		async def test_raises_not_found_when_deployment_already_tearing_down(
		self,
		state: AppDeploymentState,
		service: EdgeApplicationManagementService,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._seed_deployment(app_deployment_repo, app_instance_repo, deployment_state=state)
		with pytest.raises(NotFoundException):
		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="p"
		)

		async def test_already_terminal_instances_are_skipped(
		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,
		instance_states={
		self.INSTANCE_A: AppInstanceState.READY,
		self.INSTANCE_B: AppInstanceState.TERMINATED,
		},
		)
		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="p"
		)
		publisher.publish.assert_called_once()
		_, payload = publisher.publish.call_args.args
		assert payload["service_instance_id"] == str(self.INSTANCE_A)
		(operation,) = list(operation_repo.rows.values())
		assert operation.metadata["app_instance_ids"] == [str(self.INSTANCE_A)]
		untouched = await app_instance_repo.get_by_id(self.INSTANCE_B)
		assert untouched is not None
		assert untouched.state == AppInstanceState.TERMINATED

		async def test_no_live_instances_settles_deployment_without_publishing(
		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,
		instance_states={
		self.INSTANCE_A: AppInstanceState.TERMINATED,
		self.INSTANCE_B: AppInstanceState.TERMINATED,
		},
		)
		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="p"
		)
		publisher.publish.assert_not_called()
		assert operation_repo.rows == {}
		deployment = await app_deployment_repo.get_by_id(DEPLOYMENT_ID)
		assert deployment is not None
		assert deployment.state == AppDeploymentState.TERMINATED

		async def test_raises_when_operation_repo_unavailable(
		self,
		srm_client: AsyncMock,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		service = EdgeApplicationManagementService(
		srm_client=srm_client,
		operation_repo=None,
		app_instance_repo=app_instance_repo,
		app_deployment_repo=app_deployment_repo,
		)
		with pytest.raises(RuntimeError, match="OperationRepository is not available"):
		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="p"
		)

		async def test_raises_when_app_deployment_repo_unavailable(
		self,
		srm_client: AsyncMock,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		service = EdgeApplicationManagementService(
		srm_client=srm_client,
		operation_repo=operation_repo,
		app_instance_repo=app_instance_repo,
		app_deployment_repo=None,
		)
		with pytest.raises(RuntimeError, match="AppDeploymentRepository is not available"):
		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="p"
		)

		async def test_raises_when_publisher_unavailable(
		self,
		srm_client: AsyncMock,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> None:
		await self._seed_deployment(app_deployment_repo, app_instance_repo)
		service = EdgeApplicationManagementService(
		srm_client=srm_client,
		publisher=None,
		operation_repo=operation_repo,
		app_instance_repo=app_instance_repo,
		app_deployment_repo=app_deployment_repo,
		)
		with pytest.raises(RuntimeError, match="DataBus publisher is not available"):
		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="p"
		)

		async def test_wraps_publish_error(
		self,
		service: EdgeApplicationManagementService,
		publisher: AsyncMock,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await self._seed_deployment(app_deployment_repo, app_instance_repo)
		publisher.publish.side_effect = Exception("NATS down")
		with pytest.raises(DownstreamServiceException, match="termination"):
		await service.delete_app_deployment(
		app_deployment_id=DEPLOYMENT_ID, tenant_id="tenant-1", app_provider_id="p"
		)


		class TestHandleCompleted:
		OPERATION_ID = UUID("dddddddd-dddd-dddd-dddd-dddddddddddd")