feat: wire GET /deployments to the real service (655d1e63) · Commits · OOP / code / open-exposure-gateway

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

+9 −2

Original line number	Diff line number	Diff line
		@@ -6,6 +6,7 @@ from fastapi import APIRouter, Depends, Query, Request, Response, status
		from open_exposure_gateway.api.camara.common import IdempotencyKeyHeader
		from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas import (
		AppDeploymentId,
		AppDeploymentInfo,
		AppInstanceInfo,
		AppManifest,
		AppManifestEnvelope,
		@@ -324,19 +325,25 @@ async def create_app_deployment(
		"/deployments",
		tags=["Application"],
		summary="Retrieve a list of Application Deployment for a given App",
		response_model=list[AppDeploymentInfo],
		responses=_responses(
		BadRequestException(),
		UnauthorizedException(),
		ForbiddenException(),
		NotImplementedException(),
		DownstreamServiceException(),
		),
		)
		async def get_app_deployments(
		service: EdgeAppService,
		caller: Caller,
		appId: Annotated[Optional[UUID], Query()] = None,
		appDeploymentId: Annotated[Optional[UUID], Query()] = None,
		) -> Any:
		raise NotImplementedException(message="Multi-zone deployment is not implemented")
		return await service.get_app_deployments(
		app_id=appId,
		app_deployment_id=appDeploymentId,
		x_correlator=caller.x_correlator,
		)


		@router.delete(

tests/unit/test_eam_endpoints.py

+39 −0

Original line number	Diff line number	Diff line
		@@ -11,6 +11,7 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.router im
		)
		from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas import (
		AppDeploymentId,
		AppDeploymentInfo,
		AppInstanceInfo,
		AppInstanceStatus,
		EdgeCloudZone,
		@@ -232,3 +233,41 @@ class TestCreateAppDeployment:
		json={**_VALID_CREATE_DEPLOYMENT, "notASpecField": True},
		)
		assert response.status_code == 400


		class TestGetAppDeployments:
		def test_returns_200_with_body_shape(
		self, client: TestClient, mock_eam_service: AsyncMock
		) -> None:
		mock_eam_service.get_app_deployments.return_value = [
		AppDeploymentInfo(
		appDeploymentName="video_analytics_eu",
		appDeploymentId=_DEPLOYMENT_ID,
		appId=_APP_ID,
		edgeCloudZones=[_ZONE_ID],
		appInstances=[_INSTANCE_ID],
		)
		]
		response = client.get(f"{EAM_BASE}/deployments")
		assert response.status_code == 200
		data = response.json()
		assert len(data) == 1
		assert data[0]["appDeploymentId"] == str(_DEPLOYMENT_ID)
		assert data[0]["appId"] == str(_APP_ID)
		assert data[0]["edgeCloudZones"] == [str(_ZONE_ID)]
		assert data[0]["appInstances"] == [str(_INSTANCE_ID)]

		def test_empty_list_returns_200(self, client: TestClient, mock_eam_service: AsyncMock) -> None:
		mock_eam_service.get_app_deployments.return_value = []
		response = client.get(f"{EAM_BASE}/deployments")
		assert response.status_code == 200
		assert response.json() == []

		def test_passes_query_filters(self, client: TestClient, mock_eam_service: AsyncMock) -> None:
		mock_eam_service.get_app_deployments.return_value = []
		client.get(f"{EAM_BASE}/deployments?appId={_APP_ID}&appDeploymentId={_DEPLOYMENT_ID}")
		mock_eam_service.get_app_deployments.assert_called_once_with(
		app_id=_APP_ID,
		app_deployment_id=_DEPLOYMENT_ID,
		x_correlator=None,
		)

tests/unit/test_eam_flows.py

+16 −0

Original line number	Diff line number	Diff line
		@@ -482,6 +482,22 @@ class TestCreateAppDeploymentFlow:
		assert all(i.state == AppInstanceState.READY for i in instances)
		assert len(live_srm.instances) == 2

		def test_get_deployments_returns_populated_app_instances_after_srm_completes(
		self,
		api_client: TestClient,
		live_srm: FakeSRMClient,
		) -> None:
		create_response = api_client.post(f"{EAM_BASE}/deployments", json=CREATE_DEPLOYMENT_BODY)
		deployment_id = create_response.json()["appDeploymentId"]

		response = api_client.get(f"{EAM_BASE}/deployments?appDeploymentId={deployment_id}")
		assert response.status_code == 200
		(deployment,) = response.json()
		assert deployment["appDeploymentId"] == deployment_id
		assert deployment["appId"] == str(APP_ID)
		assert set(deployment["edgeCloudZones"]) == {str(DEPLOYMENT_ZONE_A), str(DEPLOYMENT_ZONE_B)}
		assert len(deployment["appInstances"]) == 2


		class TestCreateAppDeploymentUnregisteredAppFlow:
		"""No autouse app registration here — this class exists to prove the