feat(eam): resolve appId to app_registrations before instantiation (6a65ab28) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/edge_application_management_service.py

+7 −0

Original line number	Diff line number	Diff line
		@@ -28,6 +28,7 @@ from open_exposure_gateway.application.mappers.edge_application_mapper import (
		build_terminate_instance_command,
		)
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		DownstreamServiceException,
		NotImplementedException,
		@@ -192,6 +193,12 @@ class EdgeApplicationManagementService:
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		) -> AppInstanceInfo:
		if self._app_registration_repo is None:
		raise RuntimeError("AppRegistrationRepository is not available")
		app_registration = await self._app_registration_repo.get_by_app_id(request.appId)
		if app_registration is None:
		raise BadRequestException(message=f"App {request.appId} is not registered")

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		app_instance_id = uuid4()
		translation = build_app_deployment_translation(

tests/unit/test_eam_flows.py

+28 −0

Original line number	Diff line number	Diff line
		@@ -13,6 +13,7 @@ from typing import Any
		from unittest.mock import AsyncMock
		from uuid import uuid4

		import pytest
		from fastapi.testclient import TestClient

		from open_exposure_gateway.api.camara.edge_application_management.vwip.router import (
		@@ -64,7 +65,20 @@ def _manifest_body(num_cpu: float \| int) -> dict[str, Any]:
		}


		def _register_app(api_client: TestClient, app_id: Any = APP_ID) -> None:
		"""Registers app_id via the real POST /apps flow, so create_app_instance's
		appId -> app_registrations resolution finds it."""
		response = api_client.post(
		f"{EAM_BASE}/apps", json={**_manifest_body(num_cpu=2), "appId": str(app_id)}
		)
		assert response.status_code == 201


		class TestCreateAppInstanceFlow:
		@pytest.fixture(autouse=True)
		def _registered_app(self, api_client: TestClient) -> None:
		_register_app(api_client)

		def test_returns_202_with_location_header(
		self, api_client: TestClient, live_srm: FakeSRMClient
		) -> None:
		@@ -148,7 +162,21 @@ class TestCreateAppInstanceFlow:
		assert instances[0]["status"] == "ready"


		class TestCreateAppInstanceUnregisteredAppFlow:
		"""No autouse app registration here — this class exists to prove the
		unregistered-appId path specifically."""

		def test_returns_400_when_app_was_never_registered(self, api_client: TestClient) -> None:
		response = api_client.post(f"{EAM_BASE}/appinstances", json=CREATE_INSTANCE_BODY)
		assert response.status_code == 400
		assert response.json()["code"] == "INVALID_ARGUMENT"


		class TestDeleteAppInstanceFlow:
		@pytest.fixture(autouse=True)
		def _registered_app(self, api_client: TestClient) -> None:
		_register_app(api_client)

		def test_returns_202_and_srm_terminates_the_instance(
		self, api_client: TestClient, fake_bus: FakeDataBus, live_srm: FakeSRMClient
		) -> None:

tests/unit/test_eam_service.py

+38 −2

Original line number	Diff line number	Diff line
		@@ -16,6 +16,7 @@ from open_exposure_gateway.application.services.edge_application_management_serv
		EdgeApplicationManagementService,
		)
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		DownstreamServiceException,
		NotImplementedException,
		@@ -410,6 +411,22 @@ class TestCreateAppInstance:
		edgeCloudZoneId=ZONE_ID,
		)

		@pytest.fixture(autouse=True)
		async def _registered_app(self, app_registration_repo: FakeAppRegistrationRepository) -> None:
		# create_app_instance now resolves appId -> app_registrations before
		# anything else, so every test in this class needs the app to exist.
		await app_registration_repo.save(
		AppRegistration(
		app_registration_id=uuid4(),
		app_id=APP_ID,
		tenant_id="tenant-1",
		name="myvideoapp",
		version="1.0.0",
		package_type=PackageType.HELM,
		status=AppRegistrationStatus.REGISTERED,
		)
		)

		async def test_publishes_deploy_command(
		self, service: EdgeApplicationManagementService, publisher: AsyncMock
		) -> None:
		@@ -434,8 +451,12 @@ class TestCreateAppInstance:
		assert result.appId == APP_ID
		assert result.edgeCloudZoneId == ZONE_ID

		async def test_raises_when_publisher_unavailable(self, srm_client: AsyncMock) -> None:
		service = EdgeApplicationManagementService(srm_client=srm_client, publisher=None)
		async def test_raises_when_publisher_unavailable(
		self, srm_client: AsyncMock, app_registration_repo: FakeAppRegistrationRepository
		) -> None:
		service = EdgeApplicationManagementService(
		srm_client=srm_client, publisher=None, app_registration_repo=app_registration_repo
		)
		with pytest.raises(RuntimeError, match="DataBus publisher is not available"):
		await service.create_app_instance(
		request=self._make_request(), tenant_id="t", app_provider_id="p"
		@@ -451,6 +472,21 @@ class TestCreateAppInstance:
		)


		class TestCreateAppInstanceUnregisteredApp:
		async def test_raises_bad_request_when_app_not_registered(
		self, service: EdgeApplicationManagementService
		) -> None:
		request = CreateAppInstanceRequest(
		name="myapp_inst",
		appId=APP_ID,
		edgeCloudZoneId=ZONE_ID,
		)
		with pytest.raises(BadRequestException, match=str(APP_ID)):
		await service.create_app_instance(
		request=request, tenant_id="tenant-1", app_provider_id="provider-1"
		)


		class TestGetAppInstances:
		async def test_returns_mapped_instances(
		self, service: EdgeApplicationManagementService, srm_client: AsyncMock

Admin message