feat(eam): accept sinkCredential on deployments (03a6c4d7) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/edge_application_management_service.py

+2 −29

Changes for src/open_exposure_gateway/application/services/edge_application_management_service.py: 2 added lines, 29 removed lines.
Original line number	Diff line number	Diff line
		@@ -590,16 +590,7 @@ class EdgeApplicationManagementService:
		):
		raise RuntimeError("Operation/AppInstance/AppDeployment repositories are not available")

		# TODO: remove once callback credentials can be stored.
		if (
		request.subscriptionRequest is not None
		and request.subscriptionRequest.sinkCredential is not None
		):
		raise NotImplementedException(
		"subscriptionRequest.sinkCredential is not supported in this "
		"release: the gateway cannot yet store callback credentials, so "
		"an authenticated sink would receive no notifications"
		)
		self._require_callback_secret_key_for(request.subscriptionRequest)

		if idempotency_key is not None:
		existing_operation = await self._operation_repo.get_by_idempotency_key(
		@@ -680,25 +671,7 @@ class EdgeApplicationManagementService:
		)

		if request.subscriptionRequest is not None:
		if self._callback_registration_repo is None:
		raise RuntimeError("CallbackRegistrationRepository is not available")
		subscription = request.subscriptionRequest
		await self._callback_registration_repo.save(
		CallbackRegistration(
		id=uuid4(),
		operation_id=operation_id,
		tenant_id=tenant_id,
		api_family="edge-application-management",
		sink=subscription.sink,
		event_types=list(subscription.types or []),
		sink_credential_ref=f"secret://oeg/{operation_id}/sink-credential"
		if subscription.sinkCredential is not None
		else None,
		expires_at=subscription.config.subscriptionExpireTime
		if subscription.config
		else None,
		)
		)
		await self._save_subscription(request.subscriptionRequest, operation_id, tenant_id)

		command = build_multi_zone_deploy_command(
		app_registration_id=app_registration.app_registration_id,

tests/unit/test_eam_flows.py

+44 −0

Changes for tests/unit/test_eam_flows.py: 44 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -448,6 +448,32 @@ class TestCreateAppInstanceSinkCredentialFlow:
		},
		}

		DEPLOYMENT_BODY: dict[str, Any] = {
		**CREATE_DEPLOYMENT_BODY,
		"subscriptionRequest": {
		"sink": "https://client.example.com/callback",
		"sinkCredential": VALID_SINK_CREDENTIAL,
		"types": [
		"org.camaraproject.edge-application-management.v0.app-deployment-status-change"
		],
		},
		}

		def test_deployment_returns_501_and_saves_nothing_without_a_key(
		self,
		api_client: TestClient,
		fake_bus: FakeDataBus,
		operation_repo: FakeOperationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		) -> None:
		response = api_client.post(f"{EAM_BASE}/deployments", json=self.DEPLOYMENT_BODY)
		assert response.status_code == 501
		assert operation_repo.rows == {}
		assert app_deployment_repo.rows == {}
		assert callback_registration_repo.rows == {}
		assert fake_bus.published == []

		def test_returns_501_and_saves_nothing_without_a_key(
		self,
		api_client: TestClient,
		@@ -487,6 +513,24 @@ class TestCreateAppInstanceSinkCredentialWithKeyFlow:
		assert VALID_SINK_CREDENTIAL["accessToken"].encode() not in credential.encrypted_token
		assert callback_delivery_port.bearer_tokens == [VALID_SINK_CREDENTIAL["accessToken"]]

		def test_deployment_token_is_sent_as_bearer_on_callback(
		self,
		api_client: TestClient,
		live_srm: FakeSRMClient,
		callback_delivery_port: FakeCallbackDeliveryPort,
		callback_credential_repo: FakeCallbackCredentialRepository,
		) -> None:
		response = api_client.post(
		f"{EAM_BASE}/deployments",
		json=TestCreateAppInstanceSinkCredentialFlow.DEPLOYMENT_BODY,
		)

		assert response.status_code == 202
		(credential,) = list(callback_credential_repo.rows.values())
		assert VALID_SINK_CREDENTIAL["accessToken"].encode() not in credential.encrypted_token
		# One delivery: the per-zone CloudEvents go out as a single array.
		assert callback_delivery_port.bearer_tokens == [VALID_SINK_CREDENTIAL["accessToken"]]


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

tests/unit/test_eam_service.py

+66 −36

Changes for tests/unit/test_eam_service.py: 66 added lines, 36 removed lines.
Original line number	Diff line number	Diff line
		@@ -276,6 +276,30 @@ def service(
		)


		@pytest.fixture()
		def service_with_key(
		srm_client: AsyncMock,
		publisher: AsyncMock,
		app_registration_repo: FakeAppRegistrationRepository,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		callback_credential_repo: FakeCallbackCredentialRepository,
		) -> EdgeApplicationManagementService:
		return EdgeApplicationManagementService(
		srm_client=srm_client,
		publisher=publisher,
		app_registration_repo=app_registration_repo,
		operation_repo=operation_repo,
		app_instance_repo=app_instance_repo,
		callback_registration_repo=callback_registration_repo,
		app_deployment_repo=app_deployment_repo,
		callback_credential_repo=callback_credential_repo,
		callback_secret_key=CALLBACK_KEY,
		)


		class TestGetEdgeCloudZones:
		async def test_returns_mapped_camara_zones(
		self, service: EdgeApplicationManagementService, srm_client: AsyncMock
		@@ -1047,52 +1071,21 @@ class TestCreateAppInstance:
		assert callback.expires_at == expires_at
		assert callback.sink_credential_ref is None

		def _service_with_key(
		self,
		srm_client: AsyncMock,
		publisher: AsyncMock,
		app_registration_repo: FakeAppRegistrationRepository,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		callback_credential_repo: FakeCallbackCredentialRepository,
		) -> EdgeApplicationManagementService:
		return EdgeApplicationManagementService(
		srm_client=srm_client,
		publisher=publisher,
		app_registration_repo=app_registration_repo,
		operation_repo=operation_repo,
		app_instance_repo=app_instance_repo,
		callback_registration_repo=callback_registration_repo,
		callback_credential_repo=callback_credential_repo,
		callback_secret_key=CALLBACK_KEY,
		)

		async def test_sink_credential_is_stored_encrypted_when_key_configured(
		self,
		srm_client: AsyncMock,
		service_with_key: EdgeApplicationManagementService,
		publisher: AsyncMock,
		app_registration_repo: FakeAppRegistrationRepository,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		callback_credential_repo: FakeCallbackCredentialRepository,
		) -> None:
		callback_credential_repo = FakeCallbackCredentialRepository()
		service = self._service_with_key(
		srm_client,
		publisher,
		app_registration_repo,
		operation_repo,
		app_instance_repo,
		callback_registration_repo,
		callback_credential_repo,
		)
		request = self._make_request_with_subscription(with_credential=True)
		assert request.subscriptionRequest is not None
		sink_credential = request.subscriptionRequest.sinkCredential
		assert sink_credential is not None

		info = await service.create_app_instance(
		info = await service_with_key.create_app_instance(
		request=request, tenant_id="tenant-1", app_provider_id="provider-1"
		)

		@@ -1575,7 +1568,7 @@ class TestCreateAppDeployment:
		),
		)

		async def test_sink_credential_rejected_with_501(
		async def test_sink_credential_rejected_with_501_when_no_key(
		self, service: EdgeApplicationManagementService
		) -> None:
		with pytest.raises(NotImplementedException, match="sinkCredential"):
		@@ -1585,7 +1578,7 @@ class TestCreateAppDeployment:
		app_provider_id="provider-1",
		)

		async def test_sink_credential_rejection_persists_nothing_and_does_not_publish(
		async def test_sink_credential_rejection_without_key_persists_nothing_and_does_not_publish(
		self,
		service: EdgeApplicationManagementService,
		publisher: AsyncMock,
		@@ -1606,6 +1599,43 @@ class TestCreateAppDeployment:
		assert callback_registration_repo.rows == {}
		publisher.publish.assert_not_called()

		async def test_sink_credential_is_stored_encrypted_when_key_configured(
		self,
		service_with_key: EdgeApplicationManagementService,
		publisher: AsyncMock,
		operation_repo: FakeOperationRepository,
		app_deployment_repo: FakeAppDeploymentRepository,
		app_instance_repo: FakeAppInstanceRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		callback_credential_repo: FakeCallbackCredentialRepository,
		) -> None:
		request = self._make_request_with_subscription(with_credential=True)
		assert request.subscriptionRequest is not None
		sink_credential = request.subscriptionRequest.sinkCredential
		assert sink_credential is not None

		result = await service_with_key.create_app_deployment(
		request=request, tenant_id="tenant-1", app_provider_id="provider-1"
		)

		assert result.appDeploymentId in app_deployment_repo.rows
		(callback,) = list(callback_registration_repo.rows.values())
		assert callback.sink_credential_ref is None
		(credential,) = list(callback_credential_repo.rows.values())
		assert credential.callback_registration_id == callback.id
		assert credential.expires_at == sink_credential.accessTokenExpiresUtc
		assert decrypt_secret(credential.encrypted_token, CALLBACK_KEY) == "raw-token"
		assert b"raw-token" not in credential.encrypted_token
		saved_rows = [
		*operation_repo.rows.values(),
		*app_deployment_repo.rows.values(),
		*app_instance_repo.rows.values(),
		*callback_registration_repo.rows.values(),
		]
		for row in saved_rows:
		assert "raw-token" not in row.model_dump_json()
		publisher.publish.assert_awaited_once()

		async def test_subscription_without_credential_still_accepted(
		self, service: EdgeApplicationManagementService
		) -> None: