Loading tests/unit/test_eam_flows.py +41 −0 Original line number Diff line number Diff line Loading @@ -22,6 +22,7 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.router im ) from open_exposure_gateway.core.exceptions import DownstreamServiceException from open_exposure_gateway.domain.edge_application_management import ( AppDeploymentStatusChangeCloudEvent, AppInstanceStatusChangeCloudEvent, SRMDeployCommand, SRMTerminateCommand, Loading Loading @@ -494,6 +495,46 @@ class TestCreateAppDeploymentFlow: assert all(i.state == AppInstanceState.READY for i in instances) assert len(live_srm.instances) == 2 def test_delivers_one_deployment_status_change_array_after_srm_completes( self, api_client: TestClient, live_srm: FakeSRMClient, callback_delivery_port: FakeCallbackDeliveryPort, ) -> None: """A client subscribed to app-deployment-status-change gets exactly ONE delivery: an array of CloudEvents, one per zone, each typed app-deployment-status-change and carrying appDeploymentId -- not N separate app-instance-status-change POSTs (ADR-0008).""" body = { **CREATE_DEPLOYMENT_BODY, "subscriptionRequest": { "sink": "https://client.example.com/callback", "types": [ "org.camaraproject.edge-application-management.v0.app-deployment-status-change" ], }, } response = api_client.post(f"{EAM_BASE}/deployments", json=body) deployment_id = response.json()["appDeploymentId"] assert len(callback_delivery_port.delivered) == 1 sink, events = callback_delivery_port.delivered[0] assert sink == "https://client.example.com/callback" assert isinstance(events, list) assert len(events) == 2 assert all(isinstance(e, AppDeploymentStatusChangeCloudEvent) for e in events) assert all( e.type == "org.camaraproject.edge-application-management.v0.app-deployment-status-change" for e in events ) assert {str(e.data.appDeploymentId) for e in events} == {deployment_id} assert {str(e.data.edgeCloudZoneId) for e in events} == { str(DEPLOYMENT_ZONE_A), str(DEPLOYMENT_ZONE_B), } assert all(e.data.status == "ready" for e in events) def test_get_deployments_returns_populated_app_instances_after_srm_completes( self, api_client: TestClient, Loading tests/unit/test_eam_service.py +214 −0 Original line number Diff line number Diff line Loading @@ -31,6 +31,7 @@ from open_exposure_gateway.core.exceptions import ( NotImplementedException, ) from open_exposure_gateway.domain.edge_application_management import ( AppDeploymentStatusChangeCloudEvent, AppInstanceStatusChangeCloudEvent, SRMCapabilityRequirement, SRMCatalogPayload, Loading Loading @@ -3134,3 +3135,216 @@ class TestHandleCompleted: if isinstance(cloud_event, AppInstanceStatusChangeCloudEvent) } assert statuses == {"ready", "failed"} async def _seed_deployment_with_instances( self, app_registration_repo: FakeAppRegistrationRepository, app_instance_repo: FakeAppInstanceRepository, app_deployment_repo: FakeAppDeploymentRepository, instance_ids: list[UUID], ) -> UUID: """Seed one app_registration + one app_deployment + N instantiating instances, all under self.OPERATION_ID and DEPLOYMENT_ID, as a multi-zone POST /deployments would. Returns the resolvable app_id.""" app_id = uuid4() app_registration_id = uuid4() await app_registration_repo.save( AppRegistration( app_registration_id=app_registration_id, app_id=app_id, tenant_id="tenant-1", name="video_analytics_eu", version="1.0.0", package_type=PackageType.HELM, status=AppRegistrationStatus.REGISTERED, ) ) await app_deployment_repo.save( AppDeployment( app_deployment_id=DEPLOYMENT_ID, operation_id=self.OPERATION_ID, app_registration_id=app_registration_id, app_deployment_name="video_analytics_eu", edge_cloud_zones=[ZONE_ID], state=AppDeploymentState.INSTANTIATING, ) ) for instance_id in instance_ids: await app_instance_repo.save( AppInstance( app_instance_id=instance_id, operation_id=self.OPERATION_ID, app_registration_id=app_registration_id, edge_cloud_zone_id=ZONE_ID, state=AppInstanceState.INSTANTIATING, app_deployment_id=DEPLOYMENT_ID, ) ) return app_id async def _seed_active_deployment_callback_registration( self, callback_registration_repo: FakeCallbackRegistrationRepository ) -> CallbackRegistration: return await callback_registration_repo.save( CallbackRegistration( id=uuid4(), operation_id=self.OPERATION_ID, tenant_id="tenant-1", api_family="edge-application-management", sink="https://client.example.com/callback", event_types=[ "org.camaraproject.edge-application-management.v0.app-deployment-status-change" ], is_active=True, ) ) async def test_deployment_subscription_delivers_one_cloudevents_array( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, app_registration_repo: FakeAppRegistrationRepository, app_instance_repo: FakeAppInstanceRepository, app_deployment_repo: FakeAppDeploymentRepository, callback_registration_repo: FakeCallbackRegistrationRepository, callback_delivery_port: FakeCallbackDeliveryPort, ) -> None: """onAppDeploymentStatusChange is ONE delivery carrying an array of CloudEvents (one per instance), typed app-deployment-status-change and each carrying appDeploymentId -- not N app-instance-status-change POSTs (ADR-0008).""" ready_id = UUID("9a3f1c22-0000-4000-8000-00000000000a") failed_id = UUID("9a3f1c22-0000-4000-8000-00000000000b") await self._seed_pending_operation(operation_repo) app_id = await self._seed_deployment_with_instances( app_registration_repo, app_instance_repo, app_deployment_repo, [ready_id, failed_id], ) await self._seed_active_deployment_callback_registration(callback_registration_repo) event = SRMOperationCompleted( schema_version="1.0", operation_id=str(self.OPERATION_ID), status="partially_completed", instances=[ SRMCompletedInstance( service_instance_id=str(ready_id), zone_id=str(ZONE_ID), status="completed" ), SRMCompletedInstance( service_instance_id=str(failed_id), zone_id=str(ZONE_ID), status="failed", error={ "type": "about:blank", "title": "Zone Capacity Exceeded", "status": 503, "detail": "no capacity", }, ), ], correlation_id="corr-1", completed_at="2026-07-04T10:02:35+00:00", ) await service.handle_completed(event) assert len(callback_delivery_port.delivered) == 1 sink, events = callback_delivery_port.delivered[0] assert sink == "https://client.example.com/callback" assert isinstance(events, list) assert len(events) == 2 assert all(isinstance(e, AppDeploymentStatusChangeCloudEvent) for e in events) assert all( e.type == "org.camaraproject.edge-application-management.v0.app-deployment-status-change" for e in events ) assert all(e.data.appDeploymentId == DEPLOYMENT_ID for e in events) assert all(e.data.appId == app_id for e in events) assert {e.data.appInstanceId for e in events} == {ready_id, failed_id} assert {e.data.status for e in events} == {"ready", "failed"} async def test_deployment_whole_operation_failure_delivers_full_array( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, app_registration_repo: FakeAppRegistrationRepository, app_instance_repo: FakeAppInstanceRepository, app_deployment_repo: FakeAppDeploymentRepository, callback_registration_repo: FakeCallbackRegistrationRepository, callback_delivery_port: FakeCallbackDeliveryPort, ) -> None: """status=failed with an empty instances[] (failure before any per-zone instance completed): the fallback flips all N pre-created rows and the deployment callback still carries one array entry per instance.""" a_id = UUID("9a3f1c22-0000-4000-8000-00000000000a") b_id = UUID("9a3f1c22-0000-4000-8000-00000000000b") await self._seed_pending_operation(operation_repo) await self._seed_deployment_with_instances( app_registration_repo, app_instance_repo, app_deployment_repo, [a_id, b_id] ) await self._seed_active_deployment_callback_registration(callback_registration_repo) event = SRMOperationCompleted( schema_version="1.0", operation_id=str(self.OPERATION_ID), status="failed", instances=[], error={ "type": "about:blank", "title": "Deploy failed", "status": 500, "detail": "boom", }, correlation_id="corr-1", completed_at="2026-07-04T10:02:35+00:00", ) await service.handle_completed(event) assert len(callback_delivery_port.delivered) == 1 _, events = callback_delivery_port.delivered[0] assert isinstance(events, list) assert {e.data.appInstanceId for e in events} == {a_id, b_id} assert all(e.data.status == "failed" for e in events) async def test_deployment_array_records_one_callback_delivery_row( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, app_registration_repo: FakeAppRegistrationRepository, app_instance_repo: FakeAppInstanceRepository, app_deployment_repo: FakeAppDeploymentRepository, callback_registration_repo: FakeCallbackRegistrationRepository, callback_delivery_repo: FakeCallbackDeliveryRepository, ) -> None: """The array is one HTTP POST, hence one callback_deliveries row -- not one per instance.""" a_id = UUID("9a3f1c22-0000-4000-8000-00000000000a") b_id = UUID("9a3f1c22-0000-4000-8000-00000000000b") await self._seed_pending_operation(operation_repo) await self._seed_deployment_with_instances( app_registration_repo, app_instance_repo, app_deployment_repo, [a_id, b_id] ) await self._seed_active_deployment_callback_registration(callback_registration_repo) event = SRMOperationCompleted( schema_version="1.0", operation_id=str(self.OPERATION_ID), status="completed", instances=[ SRMCompletedInstance( service_instance_id=str(a_id), zone_id=str(ZONE_ID), status="completed" ), SRMCompletedInstance( service_instance_id=str(b_id), zone_id=str(ZONE_ID), status="completed" ), ], correlation_id="corr-1", completed_at="2026-07-04T10:02:35+00:00", ) await service.handle_completed(event) rows = list(callback_delivery_repo.rows.values()) assert len(rows) == 1 assert rows[0].state == "delivered" Loading
tests/unit/test_eam_flows.py +41 −0 Original line number Diff line number Diff line Loading @@ -22,6 +22,7 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.router im ) from open_exposure_gateway.core.exceptions import DownstreamServiceException from open_exposure_gateway.domain.edge_application_management import ( AppDeploymentStatusChangeCloudEvent, AppInstanceStatusChangeCloudEvent, SRMDeployCommand, SRMTerminateCommand, Loading Loading @@ -494,6 +495,46 @@ class TestCreateAppDeploymentFlow: assert all(i.state == AppInstanceState.READY for i in instances) assert len(live_srm.instances) == 2 def test_delivers_one_deployment_status_change_array_after_srm_completes( self, api_client: TestClient, live_srm: FakeSRMClient, callback_delivery_port: FakeCallbackDeliveryPort, ) -> None: """A client subscribed to app-deployment-status-change gets exactly ONE delivery: an array of CloudEvents, one per zone, each typed app-deployment-status-change and carrying appDeploymentId -- not N separate app-instance-status-change POSTs (ADR-0008).""" body = { **CREATE_DEPLOYMENT_BODY, "subscriptionRequest": { "sink": "https://client.example.com/callback", "types": [ "org.camaraproject.edge-application-management.v0.app-deployment-status-change" ], }, } response = api_client.post(f"{EAM_BASE}/deployments", json=body) deployment_id = response.json()["appDeploymentId"] assert len(callback_delivery_port.delivered) == 1 sink, events = callback_delivery_port.delivered[0] assert sink == "https://client.example.com/callback" assert isinstance(events, list) assert len(events) == 2 assert all(isinstance(e, AppDeploymentStatusChangeCloudEvent) for e in events) assert all( e.type == "org.camaraproject.edge-application-management.v0.app-deployment-status-change" for e in events ) assert {str(e.data.appDeploymentId) for e in events} == {deployment_id} assert {str(e.data.edgeCloudZoneId) for e in events} == { str(DEPLOYMENT_ZONE_A), str(DEPLOYMENT_ZONE_B), } assert all(e.data.status == "ready" for e in events) def test_get_deployments_returns_populated_app_instances_after_srm_completes( self, api_client: TestClient, Loading
tests/unit/test_eam_service.py +214 −0 Original line number Diff line number Diff line Loading @@ -31,6 +31,7 @@ from open_exposure_gateway.core.exceptions import ( NotImplementedException, ) from open_exposure_gateway.domain.edge_application_management import ( AppDeploymentStatusChangeCloudEvent, AppInstanceStatusChangeCloudEvent, SRMCapabilityRequirement, SRMCatalogPayload, Loading Loading @@ -3134,3 +3135,216 @@ class TestHandleCompleted: if isinstance(cloud_event, AppInstanceStatusChangeCloudEvent) } assert statuses == {"ready", "failed"} async def _seed_deployment_with_instances( self, app_registration_repo: FakeAppRegistrationRepository, app_instance_repo: FakeAppInstanceRepository, app_deployment_repo: FakeAppDeploymentRepository, instance_ids: list[UUID], ) -> UUID: """Seed one app_registration + one app_deployment + N instantiating instances, all under self.OPERATION_ID and DEPLOYMENT_ID, as a multi-zone POST /deployments would. Returns the resolvable app_id.""" app_id = uuid4() app_registration_id = uuid4() await app_registration_repo.save( AppRegistration( app_registration_id=app_registration_id, app_id=app_id, tenant_id="tenant-1", name="video_analytics_eu", version="1.0.0", package_type=PackageType.HELM, status=AppRegistrationStatus.REGISTERED, ) ) await app_deployment_repo.save( AppDeployment( app_deployment_id=DEPLOYMENT_ID, operation_id=self.OPERATION_ID, app_registration_id=app_registration_id, app_deployment_name="video_analytics_eu", edge_cloud_zones=[ZONE_ID], state=AppDeploymentState.INSTANTIATING, ) ) for instance_id in instance_ids: await app_instance_repo.save( AppInstance( app_instance_id=instance_id, operation_id=self.OPERATION_ID, app_registration_id=app_registration_id, edge_cloud_zone_id=ZONE_ID, state=AppInstanceState.INSTANTIATING, app_deployment_id=DEPLOYMENT_ID, ) ) return app_id async def _seed_active_deployment_callback_registration( self, callback_registration_repo: FakeCallbackRegistrationRepository ) -> CallbackRegistration: return await callback_registration_repo.save( CallbackRegistration( id=uuid4(), operation_id=self.OPERATION_ID, tenant_id="tenant-1", api_family="edge-application-management", sink="https://client.example.com/callback", event_types=[ "org.camaraproject.edge-application-management.v0.app-deployment-status-change" ], is_active=True, ) ) async def test_deployment_subscription_delivers_one_cloudevents_array( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, app_registration_repo: FakeAppRegistrationRepository, app_instance_repo: FakeAppInstanceRepository, app_deployment_repo: FakeAppDeploymentRepository, callback_registration_repo: FakeCallbackRegistrationRepository, callback_delivery_port: FakeCallbackDeliveryPort, ) -> None: """onAppDeploymentStatusChange is ONE delivery carrying an array of CloudEvents (one per instance), typed app-deployment-status-change and each carrying appDeploymentId -- not N app-instance-status-change POSTs (ADR-0008).""" ready_id = UUID("9a3f1c22-0000-4000-8000-00000000000a") failed_id = UUID("9a3f1c22-0000-4000-8000-00000000000b") await self._seed_pending_operation(operation_repo) app_id = await self._seed_deployment_with_instances( app_registration_repo, app_instance_repo, app_deployment_repo, [ready_id, failed_id], ) await self._seed_active_deployment_callback_registration(callback_registration_repo) event = SRMOperationCompleted( schema_version="1.0", operation_id=str(self.OPERATION_ID), status="partially_completed", instances=[ SRMCompletedInstance( service_instance_id=str(ready_id), zone_id=str(ZONE_ID), status="completed" ), SRMCompletedInstance( service_instance_id=str(failed_id), zone_id=str(ZONE_ID), status="failed", error={ "type": "about:blank", "title": "Zone Capacity Exceeded", "status": 503, "detail": "no capacity", }, ), ], correlation_id="corr-1", completed_at="2026-07-04T10:02:35+00:00", ) await service.handle_completed(event) assert len(callback_delivery_port.delivered) == 1 sink, events = callback_delivery_port.delivered[0] assert sink == "https://client.example.com/callback" assert isinstance(events, list) assert len(events) == 2 assert all(isinstance(e, AppDeploymentStatusChangeCloudEvent) for e in events) assert all( e.type == "org.camaraproject.edge-application-management.v0.app-deployment-status-change" for e in events ) assert all(e.data.appDeploymentId == DEPLOYMENT_ID for e in events) assert all(e.data.appId == app_id for e in events) assert {e.data.appInstanceId for e in events} == {ready_id, failed_id} assert {e.data.status for e in events} == {"ready", "failed"} async def test_deployment_whole_operation_failure_delivers_full_array( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, app_registration_repo: FakeAppRegistrationRepository, app_instance_repo: FakeAppInstanceRepository, app_deployment_repo: FakeAppDeploymentRepository, callback_registration_repo: FakeCallbackRegistrationRepository, callback_delivery_port: FakeCallbackDeliveryPort, ) -> None: """status=failed with an empty instances[] (failure before any per-zone instance completed): the fallback flips all N pre-created rows and the deployment callback still carries one array entry per instance.""" a_id = UUID("9a3f1c22-0000-4000-8000-00000000000a") b_id = UUID("9a3f1c22-0000-4000-8000-00000000000b") await self._seed_pending_operation(operation_repo) await self._seed_deployment_with_instances( app_registration_repo, app_instance_repo, app_deployment_repo, [a_id, b_id] ) await self._seed_active_deployment_callback_registration(callback_registration_repo) event = SRMOperationCompleted( schema_version="1.0", operation_id=str(self.OPERATION_ID), status="failed", instances=[], error={ "type": "about:blank", "title": "Deploy failed", "status": 500, "detail": "boom", }, correlation_id="corr-1", completed_at="2026-07-04T10:02:35+00:00", ) await service.handle_completed(event) assert len(callback_delivery_port.delivered) == 1 _, events = callback_delivery_port.delivered[0] assert isinstance(events, list) assert {e.data.appInstanceId for e in events} == {a_id, b_id} assert all(e.data.status == "failed" for e in events) async def test_deployment_array_records_one_callback_delivery_row( self, service: EdgeApplicationManagementService, operation_repo: FakeOperationRepository, app_registration_repo: FakeAppRegistrationRepository, app_instance_repo: FakeAppInstanceRepository, app_deployment_repo: FakeAppDeploymentRepository, callback_registration_repo: FakeCallbackRegistrationRepository, callback_delivery_repo: FakeCallbackDeliveryRepository, ) -> None: """The array is one HTTP POST, hence one callback_deliveries row -- not one per instance.""" a_id = UUID("9a3f1c22-0000-4000-8000-00000000000a") b_id = UUID("9a3f1c22-0000-4000-8000-00000000000b") await self._seed_pending_operation(operation_repo) await self._seed_deployment_with_instances( app_registration_repo, app_instance_repo, app_deployment_repo, [a_id, b_id] ) await self._seed_active_deployment_callback_registration(callback_registration_repo) event = SRMOperationCompleted( schema_version="1.0", operation_id=str(self.OPERATION_ID), status="completed", instances=[ SRMCompletedInstance( service_instance_id=str(a_id), zone_id=str(ZONE_ID), status="completed" ), SRMCompletedInstance( service_instance_id=str(b_id), zone_id=str(ZONE_ID), status="completed" ), ], correlation_id="corr-1", completed_at="2026-07-04T10:02:35+00:00", ) await service.handle_completed(event) rows = list(callback_delivery_repo.rows.values()) assert len(rows) == 1 assert rows[0].state == "delivered"
