fix(eam): wire callback delivery into the HTTP EAM service

George Papathanailrequested to merge
fix/eam-http-callback-delivery into develop

Merge request reports

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,