feat: wire POST /deployments to create_app_deployment (c74770e2) · Commits · OOP / code / open-exposure-gateway

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

+22 −2

Original line number	Diff line number	Diff line
		@@ -5,9 +5,11 @@ 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,
		AppInstanceInfo,
		AppManifest,
		AppManifestEnvelope,
		CreateAppDeploymentRequest,
		CreateAppInstanceRequest,
		EdgeCloudZone,
		EdgeCloudZoneStatus,
		@@ -286,7 +288,9 @@ async def delete_app_instance(
		@router.post(
		"/deployments",
		tags=["Application"],
		status_code=202,
		summary="Deploy an Application",
		response_model=AppDeploymentId,
		responses=_responses(
		BadRequestException(),
		UnauthorizedException(),
		@@ -296,8 +300,24 @@ async def delete_app_instance(
		DownstreamServiceException(),
		),
		)
		async def create_app_deployment() -> Any:
		raise NotImplementedException(message="Multi-zone deployment is not implemented")
		async def create_app_deployment(
		request: CreateAppDeploymentRequest,
		service: EdgeAppService,
		caller: Caller,
		http_request: Request,
		response: Response,
		idempotency_key: IdempotencyKeyHeader = None,
		) -> Any:
		deployment = await service.create_app_deployment(
		request=request,
		tenant_id=caller.tenant_id,
		app_provider_id=caller.app_provider_id,
		x_correlator=caller.x_correlator,
		idempotency_key=idempotency_key,
		)
		base = str(http_request.base_url).rstrip("/")
		response.headers["Location"] = f"{base}{BASE_PATH}/deployments/{deployment.appDeploymentId}"
		return deployment


		@router.get(

tests/unit/test_eam_endpoints.py

+64 −0

Original line number	Diff line number	Diff line
		@@ -10,6 +10,7 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.router im
		BASE_PATH as EAM_BASE,
		)
		from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas import (
		AppDeploymentId,
		AppInstanceInfo,
		AppInstanceStatus,
		EdgeCloudZone,
		@@ -52,6 +53,15 @@ _VALID_CREATE_INSTANCE: dict[str, Any] = {
		"edgeCloudZoneId": str(_ZONE_ID),
		}

		_ZONE_ID_2 = uuid4()
		_DEPLOYMENT_ID = uuid4()

		_VALID_CREATE_DEPLOYMENT: dict[str, Any] = {
		"appDeploymentName": "video_analytics_eu",
		"appId": str(_APP_ID),
		"edgeCloudZones": [str(_ZONE_ID), str(_ZONE_ID_2)],
		}


		@pytest.fixture()
		def mock_eam_service() -> AsyncMock:
		@@ -73,6 +83,7 @@ def mock_eam_service() -> AsyncMock:
		status=AppInstanceStatus.INSTANTIATING,
		edgeCloudZoneId=_ZONE_ID,
		)
		service.create_app_deployment.return_value = AppDeploymentId(appDeploymentId=_DEPLOYMENT_ID)
		return service


		@@ -168,3 +179,56 @@ class TestCreateAppInstance:
		json={"name": "myvideoapp_inst", "edgeCloudZoneId": str(_ZONE_ID)},
		)
		assert response.status_code == 400


		class TestCreateAppDeployment:
		def test_returns_202(self, client: TestClient) -> None:
		response = client.post(f"{EAM_BASE}/deployments", json=_VALID_CREATE_DEPLOYMENT)
		assert response.status_code == 202

		def test_returns_app_deployment_id_body(self, client: TestClient) -> None:
		response = client.post(f"{EAM_BASE}/deployments", json=_VALID_CREATE_DEPLOYMENT)
		assert response.json() == {"appDeploymentId": str(_DEPLOYMENT_ID)}

		def test_service_called_with_request(
		self, client: TestClient, mock_eam_service: AsyncMock
		) -> None:
		client.post(f"{EAM_BASE}/deployments", json=_VALID_CREATE_DEPLOYMENT)
		mock_eam_service.create_app_deployment.assert_called_once_with(
		request=ANY,
		tenant_id=ANY,
		app_provider_id=ANY,
		x_correlator=ANY,
		idempotency_key=ANY,
		)

		def test_location_header_set(self, client: TestClient) -> None:
		response = client.post(f"{EAM_BASE}/deployments", json=_VALID_CREATE_DEPLOYMENT)
		assert (
		response.headers["location"]
		== f"http://testserver{EAM_BASE}/deployments/{_DEPLOYMENT_ID}"
		)

		def test_missing_app_id_returns_400(self, client: TestClient) -> None:
		response = client.post(
		f"{EAM_BASE}/deployments",
		json={
		"appDeploymentName": "video_analytics_eu",
		"edgeCloudZones": [str(_ZONE_ID)],
		},
		)
		assert response.status_code == 400

		def test_empty_edge_cloud_zones_returns_400(self, client: TestClient) -> None:
		response = client.post(
		f"{EAM_BASE}/deployments",
		json={**_VALID_CREATE_DEPLOYMENT, "edgeCloudZones": []},
		)
		assert response.status_code == 400

		def test_unknown_field_returns_400(self, client: TestClient) -> None:
		response = client.post(
		f"{EAM_BASE}/deployments",
		json={**_VALID_CREATE_DEPLOYMENT, "notASpecField": True},
		)
		assert response.status_code == 400