Merge branch 'fix/eam-http-callback-delivery' into 'develop' (68ff79a6) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/core/state.py

+2 −0

Changes for src/open_exposure_gateway/core/state.py: 2 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -3,6 +3,7 @@ from typing import Protocol
		from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker

		from open_exposure_gateway.adapters.http.fm_client import FmClient
		from open_exposure_gateway.ports.callback_delivery_port import CallbackDeliveryPort
		from open_exposure_gateway.ports.databus_port import DataBusPort
		from open_exposure_gateway.ports.qod_callback_port import QodCallbackDeliveryPort
		from open_exposure_gateway.ports.srm_port import SRMClientPort
		@@ -17,6 +18,7 @@ class AppState(Protocol):
		publisher: DataBusPort
		db_engine: AsyncEngine
		session_maker: async_sessionmaker[AsyncSession]
		callback_client: CallbackDeliveryPort
		qod_callback_client: QodCallbackDeliveryPort
		traffic_influence_callback_client: TrafficInfluenceCallbackDeliveryPort
		token_verifier: TokenVerifierPort

src/open_exposure_gateway/dependencies.py

+9 −0

Changes for src/open_exposure_gateway/dependencies.py: 9 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -57,6 +57,7 @@ from open_exposure_gateway.core.exceptions import (
		UnauthorizedException,
		)
		from open_exposure_gateway.core.state import AppState
		from open_exposure_gateway.ports.callback_delivery_port import CallbackDeliveryPort
		from open_exposure_gateway.ports.database.callbacks import (
		CallbackCredentialRepository,
		CallbackDeliveryRepository,
		@@ -261,6 +262,10 @@ def get_callback_credential_repo(session: SessionDep) -> CallbackCredentialRepos
		return SqlCallbackCredentialRepository(session)


		def get_callback_client(request: Request) -> CallbackDeliveryPort:
		return get_app_state(request=request).callback_client


		def get_qod_callback_client(request: Request) -> QodCallbackDeliveryPort:
		return get_app_state(request=request).qod_callback_client

		@@ -284,6 +289,8 @@ def get_edge_app_service(
		callback_registration_repo: CallbackRegistrationRepository = Depends(
		get_callback_registration_repo
		),
		callback_delivery_repo: CallbackDeliveryRepository = Depends(get_callback_delivery_repo),
		callback_delivery_port: CallbackDeliveryPort = Depends(get_callback_client),
		app_deployment_repo: AppDeploymentRepository = Depends(get_app_deployment_repo),
		callback_credential_repo: CallbackCredentialRepository = Depends(get_callback_credential_repo),
		) -> EdgeApplicationManagementService:
		@@ -294,6 +301,8 @@ def get_edge_app_service(
		operation_repo,
		app_instance_repo,
		callback_registration_repo,
		callback_delivery_port=callback_delivery_port,
		callback_delivery_repo=callback_delivery_repo,
		app_deployment_repo=app_deployment_repo,
		callback_credential_repo=callback_credential_repo,
		callback_secret_key=get_settings().callback_settings.secret_key_bytes,

src/open_exposure_gateway/main.py

+5 −5

Changes for src/open_exposure_gateway/main.py: 5 added lines, 5 removed lines.
Original line number	Diff line number	Diff line
		@@ -564,6 +564,7 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		logger.error("Failed to initialize SRM client", error=str(e))
		raise

		callback_client = HttpCallbackClient()
		qod_callback_client = HttpQodCallbackClient()
		traffic_influence_callback_client = HttpTrafficInfluenceCallbackClient()
		fm_client = FmClient()
		@@ -587,16 +588,14 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		eam_operation_consumer = NatsOperationConsumer(
		client=publisher.client,
		subject=Subject.OPERATION_COMPLETED,
		handler=_build_operation_completed_handler(session_maker, srm_client, HttpCallbackClient()),
		handler=_build_operation_completed_handler(session_maker, srm_client, callback_client),
		)
		await eam_operation_consumer.start()

		eam_operation_status_consumer = NatsOperationStatusConsumer(
		client=publisher.client,
		subject=Subject.OPERATION_STATUS,
		handler=_build_eam_operation_status_handler(
		session_maker, srm_client, HttpCallbackClient()
		),
		handler=_build_eam_operation_status_handler(session_maker, srm_client, callback_client),
		)
		await eam_operation_status_consumer.start()
		logger.info("NATS consumer started", subject=Subject.OPERATION_STATUS)
		@@ -628,7 +627,7 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		_reconciliation_loop(
		session_maker,
		srm_client,
		HttpCallbackClient(),
		callback_client,
		interval=reconciliation_settings.interval_seconds,
		deadline=timedelta(seconds=reconciliation_settings.operation_deadline_seconds),
		publisher=publisher,
		@@ -645,6 +644,7 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
		app.state.publisher = publisher
		app.state.db_engine = db_engine
		app.state.session_maker = session_maker
		app.state.callback_client = callback_client
		app.state.qod_callback_client = qod_callback_client
		app.state.traffic_influence_callback_client = traffic_influence_callback_client
		app.state.token_verifier = token_verifier

tests/unit/test_eam_service.py

+36 −0

Changes for tests/unit/test_eam_service.py: 36 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -41,6 +41,7 @@ from open_exposure_gateway.core.exceptions import (
		NotFoundException,
		NotImplementedException,
		)
		from open_exposure_gateway.dependencies import get_edge_app_service
		from open_exposure_gateway.domain.edge_application_management import (
		AppDeploymentStatusChangeCloudEvent,
		AppInstanceStatusChangeCloudEvent,
		@@ -2604,6 +2605,41 @@ class TestDeleteAppInstance:
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)

		async def test_instance_failed_before_start_is_deleted_via_http_service(
		self,
		srm_client: AsyncMock,
		publisher: AsyncMock,
		app_registration_repo: FakeAppRegistrationRepository,
		operation_repo: FakeOperationRepository,
		app_instance_repo: FakeAppInstanceRepository,
		callback_registration_repo: FakeCallbackRegistrationRepository,
		callback_delivery_repo: FakeCallbackDeliveryRepository,
		callback_delivery_port: FakeCallbackDeliveryPort,
		app_deployment_repo: FakeAppDeploymentRepository,
		callback_credential_repo: FakeCallbackCredentialRepository,
		) -> None:
		await self._mark_failed_before_start(app_instance_repo, operation_repo)
		service = get_edge_app_service(
		srm=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_delivery_repo=callback_delivery_repo,
		callback_delivery_port=callback_delivery_port,
		app_deployment_repo=app_deployment_repo,
		callback_credential_repo=callback_credential_repo,
		)

		await service.delete_app_instance(
		app_instance_id=INSTANCE_ID, tenant_id="tenant-1", app_provider_id="VideoAppsCo"
		)

		deleted = await app_instance_repo.get_by_id(INSTANCE_ID)
		assert deleted is not None
		assert deleted.state == AppInstanceState.TERMINATED

		async def test_instance_failed_before_start_is_still_hidden_from_other_provider(
		self,
		service: EdgeApplicationManagementService,