test: add e2e flow tests for POST /deployments (4301eea7) · Commits · OOP / code / open-exposure-gateway

tests/unit/conftest.py

+8 −0

Original line number	Diff line number	Diff line
		@@ -26,6 +26,7 @@ from open_exposure_gateway.dependencies import (
		)
		from open_exposure_gateway.main import app
		from tests.unit.fakes import (
		FakeAppDeploymentRepository,
		FakeAppInstanceRepository,
		FakeAppRegistrationRepository,
		FakeCallbackDeliveryPort,
		@@ -88,6 +89,11 @@ def app_instance_repo() -> FakeAppInstanceRepository:
		return FakeAppInstanceRepository()


		@pytest.fixture()
		def app_deployment_repo() -> FakeAppDeploymentRepository:
		return FakeAppDeploymentRepository()


		@pytest.fixture()
		def callback_registration_repo() -> FakeCallbackRegistrationRepository:
		return FakeCallbackRegistrationRepository()
		@@ -146,6 +152,7 @@ def eam_service(
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		) -> EdgeApplicationManagementService:
		return EdgeApplicationManagementService(
		srm_client=fake_srm,
		@@ -154,6 +161,7 @@ def eam_service(
		operation_repo=operation_repo,
		app_instance_repo=app_instance_repo,
		callback_registration_repo=callback_registration_repo,
		app_deployment_repo=app_deployment_repo,
		)

tests/unit/test_eam_flows.py

+155 −1

Original line number	Diff line number	Diff line
		@@ -28,8 +28,14 @@ from open_exposure_gateway.domain.edge_application_management import (
		SRMZoneMetadata,
		Subject,
		)
		from open_exposure_gateway.domain.models import AppInstanceState, OperationStatus, OperationType
		from open_exposure_gateway.domain.models import (
		AppDeploymentState,
		AppInstanceState,
		OperationStatus,
		OperationType,
		)
		from tests.unit.fakes import (
		FakeAppDeploymentRepository,
		FakeAppInstanceRepository,
		FakeAppRegistrationRepository,
		FakeCallbackDeliveryPort,
		@@ -52,6 +58,15 @@ CREATE_INSTANCE_BODY: dict[str, Any] = {
		"edgeCloudZoneId": str(ZONE_ID),
		}

		DEPLOYMENT_ZONE_A = UUID("aaaaaaaa-4444-4000-8000-000000000004")
		DEPLOYMENT_ZONE_B = UUID("aaaaaaaa-5555-4000-8000-000000000005")

		CREATE_DEPLOYMENT_BODY: dict[str, Any] = {
		"appDeploymentName": "video_analytics_eu",
		"appId": str(APP_ID),
		"edgeCloudZones": [str(DEPLOYMENT_ZONE_A), str(DEPLOYMENT_ZONE_B)],
		}


		def _manifest_body(num_cpu: float \| int) -> dict[str, Any]:
		return {
		@@ -310,6 +325,145 @@ class TestCreateAppInstanceUnregisteredAppFlow:
		assert response.json()["code"] == "INVALID_ARGUMENT"


		class TestCreateAppDeploymentFlow:
		@pytest.fixture(autouse=True)
		def _registered_app(self, register_app: Callable[[Any], None]) -> None:
		register_app(APP_ID)

		def test_returns_202_with_location_header(self, api_client: TestClient) -> None:
		response = api_client.post(f"{EAM_BASE}/deployments", json=CREATE_DEPLOYMENT_BODY)
		assert response.status_code == 202
		assert response.headers["Location"].endswith(
		f"{EAM_BASE}/deployments/{response.json()['appDeploymentId']}"
		)

		def test_returns_only_app_deployment_id_body(self, api_client: TestClient) -> None:
		"""Per app-deployment-flow.md, the 202 body carries only appDeploymentId --
		the per-zone appInstanceIds are discovered later via GET /deployments, not
		returned here (unlike the single-zone /appinstances 202 body)."""
		response = api_client.post(f"{EAM_BASE}/deployments", json=CREATE_DEPLOYMENT_BODY)
		assert set(response.json().keys()) == {"appDeploymentId"}

		def test_persists_operation_app_deployment_and_app_instance_rows(
		self,
		api_client: TestClient,
		operation_repo: FakeOperationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		) -> None:
		response = api_client.post(f"{EAM_BASE}/deployments", json=CREATE_DEPLOYMENT_BODY)
		deployment_id = UUID(response.json()["appDeploymentId"])

		operations = list(operation_repo.rows.values())
		assert len(operations) == 1
		assert operations[0].status == OperationStatus.PENDING

		stored_deployment = app_deployment_repo.rows.get(deployment_id)
		assert stored_deployment is not None
		assert stored_deployment.state == AppDeploymentState.INSTANTIATING
		assert set(stored_deployment.edge_cloud_zones) == {DEPLOYMENT_ZONE_A, DEPLOYMENT_ZONE_B}

		instances = list(app_instance_repo.rows.values())
		assert len(instances) == 2
		assert {i.edge_cloud_zone_id for i in instances} == {DEPLOYMENT_ZONE_A, DEPLOYMENT_ZONE_B}
		for instance in instances:
		assert instance.app_deployment_id == deployment_id
		assert instance.state == AppInstanceState.INSTANTIATING

		def test_persists_callback_registration_when_subscription_request_present(
		self,
		api_client: TestClient,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		) -> None:
		body = {
		**CREATE_DEPLOYMENT_BODY,
		"subscriptionRequest": {
		"sink": "https://client.example.com/callback",
		"types": [
		"org.camaraproject.edge-application-management.v0.app-deployment-status-change"
		],
		},
		}
		api_client.post(f"{EAM_BASE}/deployments", json=body)

		(callback,) = list(callback_registration_repo.rows.values())
		assert callback.sink == "https://client.example.com/callback"

		def test_srm_receives_a_valid_deploy_command_with_n_targets(
		self,
		api_client: TestClient,
		fake_bus: FakeDataBus,
		app_registration_repo: FakeAppRegistrationRepository,
		) -> None:
		api_client.post(f"{EAM_BASE}/deployments", json=CREATE_DEPLOYMENT_BODY)

		deploys = [p for s, p in fake_bus.published if s == Subject.TASK_DEPLOY]
		assert len(deploys) == 1
		command = SRMDeployCommand.model_validate(deploys[0])
		registration = next(r for r in app_registration_repo.rows.values() if r.app_id == APP_ID)
		assert command.service_specification_id == str(registration.app_registration_id)
		assert len(command.targets) == 2
		assert {t.zone_id for t in command.targets} == {
		str(DEPLOYMENT_ZONE_A),
		str(DEPLOYMENT_ZONE_B),
		}
		assert command.deploy.instance_name == "video_analytics_eu"
		assert command.source == "nbi_camara"

		def test_x_correlator_header_propagates_into_command(
		self, api_client: TestClient, fake_bus: FakeDataBus
		) -> None:
		api_client.post(
		f"{EAM_BASE}/deployments",
		json=CREATE_DEPLOYMENT_BODY,
		headers={"x-correlator": "corr-123"},
		)
		_, command = fake_bus.published[0]
		assert command["correlation_id"] == "corr-123"

		def test_repeated_idempotency_key_returns_same_deployment_and_does_not_republish(
		self, api_client: TestClient, fake_bus: FakeDataBus
		) -> None:
		headers = {"Idempotency-Key": "client-retry-1"}
		first = api_client.post(
		f"{EAM_BASE}/deployments", json=CREATE_DEPLOYMENT_BODY, headers=headers
		)
		second = api_client.post(
		f"{EAM_BASE}/deployments", json=CREATE_DEPLOYMENT_BODY, headers=headers
		)
		assert first.json()["appDeploymentId"] == second.json()["appDeploymentId"]
		deploys = [p for s, p in fake_bus.published if s == Subject.TASK_DEPLOY]
		assert len(deploys) == 1

		def test_publish_failure_maps_to_503_envelope(
		self, api_client: TestClient, fake_bus: FakeDataBus
		) -> None:
		fake_bus.publish = AsyncMock( # type: ignore[method-assign]
		side_effect=RuntimeError("nats down")
		)
		response = api_client.post(f"{EAM_BASE}/deployments", json=CREATE_DEPLOYMENT_BODY)
		assert response.status_code == 503
		assert response.json()["code"] == "UNAVAILABLE"

		def test_cluster_refs_longer_than_zones_returns_400(self, api_client: TestClient) -> None:
		body = {
		**CREATE_DEPLOYMENT_BODY,
		"kubernetesClusterRefs": [str(uuid4()), str(uuid4()), str(uuid4())],
		}
		response = api_client.post(f"{EAM_BASE}/deployments", json=body)
		assert response.status_code == 400


		class TestCreateAppDeploymentUnregisteredAppFlow:
		"""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}/deployments", json=CREATE_DEPLOYMENT_BODY)
		assert response.status_code == 400
		assert response.json()["code"] == "INVALID_ARGUMENT"


		class TestDeleteAppInstanceFlow:
		@pytest.fixture(autouse=True)
		def _registered_app(self, register_app: Callable[[Any], None]) -> None: