Commit ac1f594a authored by George Papathanail's avatar George Papathanail
Browse files

feat: stop stop notifying a sink that answer 410 gone

parent 08ad57cb

src/open_exposure_gateway/adapters/http/qod_callback_client.py

+3 −0
Changes for src/open_exposure_gateway/adapters/http/qod_callback_client.py: 3 added lines, 0 removed lines.
Original line number Diff line number Diff line 
import httpx



from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.domain.errors import CallbackSinkGoneError

from open_exposure_gateway.domain.quality_on_demand import QosStatusChangedCloudEvent
@@ -23,4 +24,6 @@ class HttpQodCallbackClient: 
            response = await client.post(

                sink, content=event.model_dump_json(exclude_none=True), headers=headers

            )

        if response.status_code == httpx.codes.GONE:

            raise CallbackSinkGoneError(f"sink answered {response.status_code}")

        response.raise_for_status()

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+14 −0
Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 14 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -40,6 +40,7 @@ from open_exposure_gateway.domain.errors import ( 
    CallbackCredentialExpiredError,

    CallbackCredentialRevokedError,

    CallbackSecretError,

    CallbackSinkGoneError,

)

from open_exposure_gateway.domain.models import (

    CallbackCredential,
@@ -372,6 +373,19 @@ class QualityOnDemandService: 
                registration.sink, cloud_event, correlation_id, bearer_token=bearer_token

            )

            state = "delivered"

        except CallbackSinkGoneError as exc:

            state = "gone"

            last_error = str(exc)

            logger.info(

                "qod_callback_sink_gone",

                callback_registration_id=str(registration.id),

                sink=registration.sink,

            )

            await self._callback_registration_repo.save(

                registration.model_copy(update={"is_active": False})

            )

            if self._callback_credential_repo is not None:

                await self._callback_credential_repo.revoke_by_registration_id(registration.id)

        except Exception as exc:

            state = "failed"

            last_error = str(exc)

src/open_exposure_gateway/domain/errors.py

+4 −0
Changes for src/open_exposure_gateway/domain/errors.py: 4 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -8,3 +8,7 @@ class CallbackCredentialExpiredError(CallbackSecretError): 


class CallbackCredentialRevokedError(CallbackSecretError):

    """The client's callback token was wiped; skip the delivery, never send it unauthenticated."""





class CallbackSinkGoneError(Exception):

    """The client's sink answered 410 Gone; stop notifying it."""

tests/unit/fakes.py

+3 −0
Changes for tests/unit/fakes.py: 3 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -710,6 +710,7 @@ class FakeQodCallbackDeliveryPort: 
        self.delivered: list[tuple[str, QosStatusChangedCloudEvent]] = []

        self.correlators: list[str] = []

        self.bearer_tokens: list[str | None] = []

        self.error: Exception | None = None



    async def deliver(

        self,
@@ -718,6 +719,8 @@ class FakeQodCallbackDeliveryPort: 
        x_correlator: str,

        bearer_token: str | None = None,

    ) -> None:

        if self.error is not None:

            raise self.error

        self.delivered.append((sink, event))

        self.correlators.append(x_correlator)

        self.bearer_tokens.append(bearer_token)

tests/unit/test_qod_callback_client.py

+19 −0
Changes for tests/unit/test_qod_callback_client.py: 19 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -8,6 +8,7 @@ import httpx 
import pytest



from open_exposure_gateway.adapters.http.qod_callback_client import HttpQodCallbackClient

from open_exposure_gateway.domain.errors import CallbackSinkGoneError

from open_exposure_gateway.domain.quality_on_demand import (

    EventQosStatusChangedData,

    QosStatusChangedCloudEvent,
@@ -102,3 +103,21 @@ async def test_error_response_raises_http_status_error(monkeypatch: pytest.Monke 
        await client.deliver(SINK, EVENT, CORRELATOR, bearer_token="client-issued-sink-token")



    assert exc_info.value.response.status_code == 401





async def test_gone_response_raises_sink_gone_error(monkeypatch: pytest.MonkeyPatch) -> None:

    client = _client(monkeypatch, lambda request: httpx.Response(410))



    with pytest.raises(CallbackSinkGoneError):

        await client.deliver(SINK, EVENT, CORRELATOR, bearer_token="client-issued-sink-token")





async def test_server_error_still_raises_http_status_error(

    monkeypatch: pytest.MonkeyPatch,

) -> None:

    client = _client(monkeypatch, lambda request: httpx.Response(500))



    with pytest.raises(httpx.HTTPStatusError) as exc_info:

        await client.deliver(SINK, EVENT, CORRELATOR)



    assert exc_info.value.response.status_code == 500