feat: persist operations+app_instances before terminate publish (f224df61) · Commits · OOP / code / open-exposure-gateway

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

+1 −0

Original line number	Diff line number	Diff line
		@@ -224,6 +224,7 @@ async def delete_app_instance(
		) -> Response:
		await service.delete_app_instance(
		app_instance_id=appInstanceId,
		tenant_id=caller.tenant_id,
		app_provider_id=caller.app_provider_id,
		x_correlator=caller.x_correlator,
		)

src/open_exposure_gateway/application/services/edge_application_management_service.py

+32 −1

Original line number	Diff line number	Diff line
		@@ -32,6 +32,7 @@ from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		DownstreamServiceException,
		NotFoundException,
		NotImplementedException,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		@@ -354,8 +355,21 @@ class EdgeApplicationManagementService:
		return [build_app_instance_info(i) for i in instances]

		async def delete_app_instance(
		self, app_instance_id: UUID, app_provider_id: str, x_correlator: Optional[str] = None
		self,
		app_instance_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")

		app_instance = await self._app_instance_repo.get_by_id(app_instance_id)
		if app_instance is None:
		raise NotFoundException(message=f"App instance {app_instance_id} not found")

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		command = build_terminate_instance_command(
		app_instance_id=app_instance_id,
		@@ -364,6 +378,23 @@ class EdgeApplicationManagementService:
		correlation_id=correlation_id,
		requested_at=requested_at,
		)

		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_instance.app_registration_id,
		)
		)
		await self._app_instance_repo.save(
		app_instance.model_copy(update={"state": AppInstanceState.TERMINATING})
		)

		await self._publish(
		Subject.TASK_TERMINATE,
		command,

tests/unit/test_eam_flows.py

+33 −1

Original line number	Diff line number	Diff line
		@@ -27,7 +27,7 @@ from open_exposure_gateway.domain.edge_application_management import (
		SRMTerminateCommand,
		Subject,
		)
		from open_exposure_gateway.domain.models import AppInstanceState, OperationStatus
		from open_exposure_gateway.domain.models import AppInstanceState, OperationStatus, OperationType
		from tests.unit.fakes import (
		FakeAppInstanceRepository,
		FakeCallbackDeliveryPort,
		@@ -294,6 +294,38 @@ class TestDeleteAppInstanceFlow:
		assert command.service_specification_id is None
		assert srm_id not in live_srm.instances

		def test_returns_404_for_unknown_instance(self, api_client: TestClient) -> None:
		response = api_client.delete(f"{EAM_BASE}/appinstances/{uuid4()}")
		assert response.status_code == 404

		def test_persists_pending_terminate_operation_and_terminating_state(
		self,
		api_client: TestClient,
		fake_bus: FakeDataBus,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		"""Uses plain api_client (not live_srm) so nothing auto-completes the
		terminate -- this test is about the request-side PENDING/terminating
		writes, not the completion path."""
		create_response = api_client.post(f"{EAM_BASE}/appinstances", json=CREATE_INSTANCE_BODY)
		instance_id = create_response.json()["appInstanceId"]

		response = api_client.delete(f"{EAM_BASE}/appinstances/{instance_id}")

		assert response.status_code == 202
		terminate_operations = [
		op
		for op in operation_repo.rows.values()
		if op.operation_type == OperationType.TERMINATE
		]
		assert len(terminate_operations) == 1
		assert terminate_operations[0].status == OperationStatus.PENDING

		updated_instance = app_instance_repo.rows.get(UUID(instance_id))
		assert updated_instance is not None
		assert updated_instance.state == AppInstanceState.TERMINATING


		class TestSubmitAppFlow:
		def test_returns_201_and_registers_catalog_entry_in_srm(

tests/unit/test_eam_service.py

+83 −5

Original line number	Diff line number	Diff line
		@@ -22,6 +22,7 @@ from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		DownstreamServiceException,
		NotFoundException,
		NotImplementedException,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		@@ -806,11 +807,26 @@ class TestGetAppInstances:


		class TestDeleteAppInstance:
		@pytest.fixture(autouse=True)
		async def _seeded_app_instance(self, app_instance_repo: FakeAppInstanceRepository) -> None:
		await app_instance_repo.save(
		AppInstance(
		app_instance_id=INSTANCE_ID,
		operation_id=uuid4(),
		app_registration_id=uuid4(),
		edge_cloud_zone_id=ZONE_ID,
		state=AppInstanceState.READY,
		)
		)

		async def test_publishes_terminate_command(
		self, service: EdgeApplicationManagementService, publisher: AsyncMock
		) -> None:
		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, app_provider_id="VideoAppsCo", x_correlator="corr-1"
		app_instance_id=INSTANCE_ID,
		tenant_id="tenant-1",
		app_provider_id="VideoAppsCo",
		x_correlator="corr-1",
		)
		publisher.publish.assert_called_once()
		subject, payload = publisher.publish.call_args.args
		@@ -819,17 +835,79 @@ class TestDeleteAppInstance:
		assert payload["app_provider_id"] == "VideoAppsCo"
		assert payload["correlation_id"] == "corr-1"

		async def test_raises_when_publisher_unavailable(self, srm_client: AsyncMock) -> None:
		service = EdgeApplicationManagementService(srm_client=srm_client, publisher=None)
		async def test_persists_pending_operation_and_terminating_state(
		self,
		service: EdgeApplicationManagementService,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		await service.delete_app_instance(
		app_instance_id=INSTANCE_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

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

		async def test_raises_not_found_for_unknown_instance(
		self, service: EdgeApplicationManagementService
		) -> None:
		with pytest.raises(NotFoundException):
		await service.delete_app_instance(
		app_instance_id=uuid4(), 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,
		) -> None:
		service = EdgeApplicationManagementService(
		srm_client=srm_client,
		publisher=None,
		operation_repo=operation_repo,
		app_instance_repo=app_instance_repo,
		)
		with pytest.raises(RuntimeError, match="DataBus publisher is not available"):
		await service.delete_app_instance(app_instance_id=INSTANCE_ID, app_provider_id="p")
		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="p"
		)

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

		async def test_raises_when_app_instance_repo_unavailable(
		self, srm_client: AsyncMock, operation_repo: FakeOperationRepository
		) -> None:
		service = EdgeApplicationManagementService(
		srm_client=srm_client, operation_repo=operation_repo, app_instance_repo=None
		)
		with pytest.raises(RuntimeError, match="AppInstanceRepository is not available"):
		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="p"
		)

		async def test_wraps_publish_error(
		self, service: EdgeApplicationManagementService, publisher: AsyncMock
		) -> None:
		publisher.publish.side_effect = Exception("NATS down")
		with pytest.raises(DownstreamServiceException, match="termination"):
		await service.delete_app_instance(app_instance_id=INSTANCE_ID, app_provider_id="p")
		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="p"
		)


		class TestHandleCompleted:

Admin message