diff --git a/docs/docs/en/getting-started/exception-handlers/index.md b/docs/docs/en/getting-started/exception-handlers/index.md new file mode 100644 index 0000000000..bf74037bf4 --- /dev/null +++ b/docs/docs/en/getting-started/exception-handlers/index.md @@ -0,0 +1,61 @@ +--- +description: >- + Handle AIOKafka message processing errors with subscriber, broker, + and FastStream exception handlers. +--- + +# Exception Handlers + +With AIOKafka, you can register an `exception_handler` on a subscriber, +on the broker, or on the `FastStream` application. + +Handlers can be synchronous or asynchronous. Each handler receives the +exception and returns a boolean. + +## Handler Order + +FastStream calls the handlers in this order: + +1. The subscriber's handler. +2. The broker's handler. +3. The application's handler. + +Missing handlers are skipped. Processing stops when a handler returns `True`. + +## Return Value + +- `True`: the exception is handled and suppressed. +- `False`: processing continues with the next handler. + +If no handler returns `True`, the original exception is raised again. + +## Example + +```python linenums="1" +{!> docs_src/getting_started/exception_handlers/kafka.py !} +``` + +Publish one of these messages to the `exception-handlers` topic: + +| Message | Exception | Result | +|---------|-----------|--------| +| `subscriber` | `ValueError` | Suppressed by the subscriber's handler | +| `broker` | `RuntimeError` | Suppressed by the broker's handler | +| `application` | `TypeError` | Logged by the application's handler and raised again | + +## Context + +Exception handlers support [context injection](../context.md){.internal-link}. +The application handler in the example uses `Logger` to access the logger +for the current message. + +## Middleware and Acknowledgement + +If middleware has already suppressed an exception, the exception handlers +are not called. + +Returning `True` controls exception propagation. Message acknowledgement +follows the broker's [acknowledgement policy](../acknowledgement.md){.internal-link}. + +`StopConsume` and `StopApplication` continue through the normal shutdown +handling and are not passed to these callbacks. \ No newline at end of file diff --git a/docs/docs/navigation_template.txt b/docs/docs/navigation_template.txt index 4d114ad1b4..d9eba0a2bc 100644 --- a/docs/docs/navigation_template.txt +++ b/docs/docs/navigation_template.txt @@ -22,6 +22,7 @@ search: - [Routers](getting-started/routers/index.md) - [Dependencies](getting-started/dependencies/index.md) - [Context](getting-started/context.md) + - [Exception Handlers](getting-started/exception-handlers/index.md) - [Custom Serialization](getting-started/serialization/index.md) - [Parser](getting-started/serialization/parser.md) - [Decoder](getting-started/serialization/decoder.md) diff --git a/docs/docs_src/getting_started/exception_handlers/__init__.py b/docs/docs_src/getting_started/exception_handlers/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/docs/docs_src/getting_started/exception_handlers/kafka.py b/docs/docs_src/getting_started/exception_handlers/kafka.py new file mode 100644 index 0000000000..b31c19e3a8 --- /dev/null +++ b/docs/docs_src/getting_started/exception_handlers/kafka.py @@ -0,0 +1,37 @@ +from faststream import FastStream, Logger +from faststream.kafka import KafkaBroker + + +def subscriber_exception_handler(exc: BaseException) -> bool: + return isinstance(exc, ValueError) + + +async def broker_exception_handler(exc: BaseException) -> bool: + return isinstance(exc, RuntimeError) + + +async def app_exception_handler( + exc: BaseException, + logger: Logger, +) -> bool: + logger.error("Unhandled message processing error", exc_info=exc) + return False + + +broker = KafkaBroker( + "localhost:9092", + exception_handler=broker_exception_handler, +) +app = FastStream(broker, exception_handler=app_exception_handler) + + +@broker.subscriber( + "exception-handlers", + exception_handler=subscriber_exception_handler, +) +async def handle(message: str) -> None: + if message == "subscriber": + raise ValueError("subscriber error") + if message == "broker": + raise RuntimeError("broker error") + raise TypeError("application error") diff --git a/faststream/_internal/broker/broker.py b/faststream/_internal/broker/broker.py index cf2b3efe79..4b4ece2445 100644 --- a/faststream/_internal/broker/broker.py +++ b/faststream/_internal/broker/broker.py @@ -54,6 +54,8 @@ def __init__( routers=routers, config=config, ) + self._exception_handler = config.broker_exception_handler + self.specification = specification self.running = False @@ -93,6 +95,9 @@ def _update_fd_config(self, config: "FastDependsConfig") -> None: """Private method to change broker config state by outer application.""" self.config.fd_config = config | self.config.fd_config + for subscriber in self.subscribers: + subscriber._setup_exception_handlers() + async def start(self) -> None: # TODO: filter by already running handlers after TestClient refactor for sub in self.subscribers: diff --git a/faststream/_internal/broker/registrator.py b/faststream/_internal/broker/registrator.py index 7704be0da5..53532cf3a0 100644 --- a/faststream/_internal/broker/registrator.py +++ b/faststream/_internal/broker/registrator.py @@ -114,6 +114,9 @@ def include_router( router.config.add_config(self.config) self.routers.append(router) + for subscriber in router.subscribers: + subscriber._setup_exception_handlers() + @property def parent(self) -> "Registrator[MsgType, Any] | None": return self.__parent diff --git a/faststream/_internal/configs/broker.py b/faststream/_internal/configs/broker.py index 667b78140c..d8afbd99df 100644 --- a/faststream/_internal/configs/broker.py +++ b/faststream/_internal/configs/broker.py @@ -9,6 +9,7 @@ from faststream._internal.logger import LoggerState from faststream._internal.producer import ProducerProto, ProducerUnset from faststream._internal.types import IdGenerator +from faststream._internal.utils import apply_types from faststream.message import gen_cor_id if TYPE_CHECKING: @@ -16,7 +17,11 @@ from faststream._internal.context import ContextRepo from faststream._internal.parser import CodecProto - from faststream._internal.types import BrokerMiddleware, CustomCallable + from faststream._internal.types import ( + AsyncExceptionHandler, + BrokerMiddleware, + CustomCallable, + ) from faststream.middlewares import AckPolicy @@ -47,6 +52,7 @@ class BrokerConfig: broker_parser: Optional["CustomCallable"] = None broker_decoder: Optional["CustomCallable"] = None broker_codec: Optional["CodecProto"] = None + broker_exception_handler: Optional["AsyncExceptionHandler"] = None producer: "ProducerProto[Any]" = field(default_factory=ProducerUnset) logger: "LoggerState" = field(default_factory=LoggerState) @@ -67,6 +73,13 @@ def __post_init__(self) -> None: # untyped callers still pass a generator: the first subscriber would spend it self.broker_dependencies = tuple(self.broker_dependencies) + if self.broker_exception_handler is not None: + self.broker_exception_handler = apply_types( + self.broker_exception_handler, + serializer_cls=self.fd_config._serializer, + context__=self.context, + ) + @property def resolved_underlying_driver_annotations( self, diff --git a/faststream/_internal/configs/endpoint.py b/faststream/_internal/configs/endpoint.py index 6beaf99d54..09f487d5db 100644 --- a/faststream/_internal/configs/endpoint.py +++ b/faststream/_internal/configs/endpoint.py @@ -2,10 +2,11 @@ from typing import TYPE_CHECKING from faststream._internal.constants import EMPTY +from faststream._internal.utils import apply_types from faststream.middlewares import AckPolicy if TYPE_CHECKING: - from faststream._internal.types import AsyncCallable + from faststream._internal.types import AsyncCallable, AsyncExceptionHandler from .broker import BrokerConfig @@ -23,12 +24,21 @@ class PublisherUsecaseConfig(EndpointConfig): @dataclass(kw_only=True) class SubscriberUsecaseConfig(EndpointConfig): no_reply: bool = False + exception_handler: "AsyncExceptionHandler | None" = None _ack_policy: AckPolicy = field(default_factory=lambda: EMPTY, repr=False) parser: "AsyncCallable" = field(init=False) decoder: "AsyncCallable" = field(init=False) + def __post_init__(self) -> None: + if self.exception_handler is not None: + self.exception_handler = apply_types( + self.exception_handler, + serializer_cls=self._outer_config.fd_config._serializer, + context__=self._outer_config.context, + ) + @property def auto_ack_disabled(self) -> bool: return self.ack_policy is AckPolicy.MANUAL diff --git a/faststream/_internal/endpoint/subscriber/usecase.py b/faststream/_internal/endpoint/subscriber/usecase.py index 2bc7ca08bc..0127915ebf 100644 --- a/faststream/_internal/endpoint/subscriber/usecase.py +++ b/faststream/_internal/endpoint/subscriber/usecase.py @@ -27,9 +27,9 @@ P_HandlerParams, T_HandlerReturn, ) -from faststream._internal.utils.functions import to_async +from faststream._internal.utils import apply_types, to_async from faststream.exceptions import StopConsume, SubscriberNotFound -from faststream.middlewares import AcknowledgementMiddleware +from faststream.middlewares import AcknowledgementMiddleware, ExceptionMiddleware from faststream.middlewares.logging import CriticalLogMiddleware from faststream.response import ensure_response @@ -48,6 +48,7 @@ from faststream._internal.endpoint.publisher import PublisherProto from faststream._internal.parser import CodecProto from faststream._internal.types import ( + AsyncExceptionHandler, AsyncFilter, BrokerMiddleware, CustomCallable, @@ -76,6 +77,8 @@ class SubscriberUsecase(Endpoint, Generic[MsgType]): "_call_decorators", "_call_options", "_decoder", + "_exception_handlers", + "_exception_middleware", "_no_reply", "_parser", "ack_policy", @@ -85,6 +88,8 @@ class SubscriberUsecase(Endpoint, Generic[MsgType]): "specification", ) + _exception_handlers: tuple["AsyncExceptionHandler", ...] + lock: "AbstractContextManager[Any]" graceful_timeout: float | None @@ -103,6 +108,7 @@ def __init__( self._no_reply = config.no_reply self._parser = config.parser self._decoder = config.decoder + self._exception_handler = config.exception_handler self.ack_policy = config.ack_policy self.__auto_ack_disabled = config.auto_ack_disabled @@ -119,6 +125,11 @@ def __init__( self.running = False self.lock = nullcontext() + self._setup_exception_handlers() + self._exception_middleware = ExceptionMiddleware( + handlers={Exception: self._handle_exception}, + ) + @property def _broker_middlewares(self) -> Sequence["BrokerMiddleware[MsgType]"]: return self._outer_config.broker_middlewares @@ -338,6 +349,39 @@ def real_wrapper( return real_wrapper(func) + @property + def _app_exception_handler(self) -> "AsyncExceptionHandler | None": + app_handler: AsyncExceptionHandler | None = getattr( + self._outer_config.context.get("app"), + "_exception_handler", + None, + ) + return app_handler + + def _setup_exception_handlers(self) -> None: + config = self._outer_config.fd_config + self._exception_handlers = tuple( + apply_types( + handler, + serializer_cls=config._serializer, + dependency_provider=config.provider, + context__=config.context, + ) + for handler in ( + self._exception_handler, + self._outer_config.broker_exception_handler, + self._app_exception_handler, + ) + if handler is not None + ) + + async def _handle_exception(self, exc: Exception) -> None: + for handler in self._exception_handlers: + if await handler(exc): + return + + raise exc + async def consume(self, msg: MsgType) -> Any: """Consume a message asynchronously.""" if not self.running: @@ -451,10 +495,12 @@ async def process_message(self, msg: MsgType) -> "Response": def __build__middlewares_stack(self) -> tuple["BrokerMiddleware[MsgType]", ...]: logger_state = self._outer_config.logger + broker_middlewares: tuple[BrokerMiddleware[MsgType], ...] if self.__auto_ack_disabled: broker_middlewares = ( CriticalLogMiddleware(logger_state), + self._exception_middleware, *self._broker_middlewares, ) @@ -466,6 +512,7 @@ def __build__middlewares_stack(self) -> tuple["BrokerMiddleware[MsgType]", ...]: fallback_parser=self._parser, ), CriticalLogMiddleware(logger_state), + self._exception_middleware, *self._broker_middlewares, ) diff --git a/faststream/_internal/types.py b/faststream/_internal/types.py index e07df2b639..379897c23e 100644 --- a/faststream/_internal/types.py +++ b/faststream/_internal/types.py @@ -43,6 +43,10 @@ AsyncFilter: TypeAlias = Callable[[StreamMsg], Awaitable[bool]] Filter: TypeAlias = SyncFilter[StreamMsg] | AsyncFilter[StreamMsg] +SyncExceptionHandler: TypeAlias = Callable[..., bool] +AsyncExceptionHandler: TypeAlias = Callable[..., Awaitable[bool]] +ExceptionHandler: TypeAlias = SyncExceptionHandler | AsyncExceptionHandler + SyncCallable: TypeAlias = Callable[ [Any], Any, diff --git a/faststream/_internal/utils/__init__.py b/faststream/_internal/utils/__init__.py index 061216397e..2948d7cced 100644 --- a/faststream/_internal/utils/__init__.py +++ b/faststream/_internal/utils/__init__.py @@ -1,8 +1,9 @@ from fast_depends import inject as apply_types -from .functions import to_async +from .functions import ensure_exception_handler, to_async __all__ = ( "apply_types", + "ensure_exception_handler", "to_async", ) diff --git a/faststream/_internal/utils/functions.py b/faststream/_internal/utils/functions.py index 97b92e4f1b..09e5d40cde 100644 --- a/faststream/_internal/utils/functions.py +++ b/faststream/_internal/utils/functions.py @@ -3,6 +3,7 @@ from concurrent.futures import Executor from functools import partial, wraps from typing import ( + TYPE_CHECKING, Any, TypeVar, cast, @@ -18,8 +19,12 @@ from faststream._internal.basic_types import F_Return, F_Spec +if TYPE_CHECKING: + from faststream._internal.types import AsyncExceptionHandler, ExceptionHandler + __all__ = ( "call_or_await", + "ensure_exception_handler", "to_async", ) @@ -56,6 +61,15 @@ async def to_async_wrapper(*args: F_Spec.args, **kwargs: F_Spec.kwargs) -> F_Ret return to_async_wrapper +def ensure_exception_handler( + handler: "ExceptionHandler | None", +) -> "AsyncExceptionHandler | None": + if handler is None: + return None + + return to_async(handler) + + async def return_input(x: Any) -> Any: return x diff --git a/faststream/app.py b/faststream/app.py index 47951ad512..606267942f 100644 --- a/faststream/app.py +++ b/faststream/app.py @@ -19,6 +19,7 @@ from faststream._internal.context import ContextRepo from faststream._internal.di import FastDependsConfig from faststream._internal.logger import logger +from faststream._internal.utils import apply_types, to_async from faststream.asgi.app import AsgiFastStream if TYPE_CHECKING: @@ -31,6 +32,7 @@ SettingField, ) from faststream._internal.broker import BrokerUsecase + from faststream._internal.types import AsyncExceptionHandler, ExceptionHandler from faststream.asgi import AsyncAPIRoute from faststream.asgi.types import ASGIApp from faststream.specification.base import SpecificationFactory @@ -49,6 +51,7 @@ def __init__( provider: Optional["Provider"] = None, serializer: Optional["SerializerProto"] = EMPTY, context: ContextRepo | None = None, + exception_handler: "ExceptionHandler | None" = None, lifespan: Optional["Lifespan"] = None, on_startup: Sequence["AnyCallable"] = (), after_startup: Sequence["AnyCallable"] = (), @@ -72,8 +75,21 @@ def __init__( specification=specification, ) + if exception_handler is not None: + self._exception_handler: AsyncExceptionHandler | None = apply_types( + to_async(exception_handler), + serializer_cls=self.config._serializer, + ) + else: + self._exception_handler = None + self._should_exit = False + # super().__init__ registers brokers before the application handler exists. + for broker in self.brokers: + for subscriber in broker.subscribers: + subscriber._setup_exception_handlers() + async def run( self, log_level: int = logging.INFO, diff --git a/faststream/confluent/subscriber/config.py b/faststream/confluent/subscriber/config.py index 729c65c90b..ae3f9f8ca1 100644 --- a/faststream/confluent/subscriber/config.py +++ b/faststream/confluent/subscriber/config.py @@ -31,6 +31,7 @@ class KafkaSubscriberConfig(SubscriberUsecaseConfig): connection_data: dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: + super().__post_init__() self.connection_data["enable_auto_commit"] = self.ack_first @property diff --git a/faststream/kafka/broker/broker.py b/faststream/kafka/broker/broker.py index 6b70610019..b414ac8d5f 100644 --- a/faststream/kafka/broker/broker.py +++ b/faststream/kafka/broker/broker.py @@ -26,6 +26,7 @@ from faststream._internal.context.repository import ContextRepo from faststream._internal.di import FastDependsConfig from faststream._internal.types import IdGenerator +from faststream._internal.utils import ensure_exception_handler from faststream._internal.utils.data import filter_by_dict from faststream.exceptions import IncorrectState from faststream.kafka._compat import AIOKAFKA_V013, validate_client_rack @@ -61,6 +62,7 @@ from faststream._internal.types import ( BrokerMiddleware, CustomCallable, + ExceptionHandler, ) from faststream.kafka.message import KafkaMessage from faststream.kafka.types import KafkaSendableMessage @@ -238,6 +240,7 @@ def __init__( decoder: Optional["CustomCallable"] = None, codec: Optional["CodecProto"] = None, parser: Optional["CustomCallable"] = None, + exception_handler: "ExceptionHandler | None" = None, dependencies: Sequence["Dependant"] = (), middlewares: Sequence["BrokerMiddleware[Any, Any]"] = (), routers: Iterable[KafkaRegistrator] = (), @@ -359,6 +362,10 @@ def __init__( Custom codec object. parser (Optional[CustomCallable]): Custom parser object. + exception_handler (ExceptionHandler | None): + Sync or async exception handler called when a subscriber's + exception handler is absent or returns False. + Return True if handled, otherwise False for default handling. dependencies (Sequence[Dependant]): Dependencies to apply to all broker subscribers. middlewares (Sequence[BrokerMiddlewarep[Any, Any]]): @@ -474,6 +481,7 @@ def __init__( broker_decoder=decoder, broker_codec=codec, broker_parser=parser, + broker_exception_handler=ensure_exception_handler(exception_handler), broker_middlewares=middlewares, logger=make_kafka_logger_state( logger=logger, diff --git a/faststream/kafka/broker/registrator.py b/faststream/kafka/broker/registrator.py index 125d27f88f..ceca1222b2 100644 --- a/faststream/kafka/broker/registrator.py +++ b/faststream/kafka/broker/registrator.py @@ -30,6 +30,7 @@ from faststream._internal.types import ( BrokerMiddleware, CustomCallable, + ExceptionHandler, ) from faststream.kafka.publisher.usecase import ( BatchPublisher, @@ -96,6 +97,7 @@ def subscriber( max_workers: None = None, ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -146,6 +148,7 @@ def subscriber( max_workers: None = None, ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -196,6 +199,7 @@ def subscriber( max_workers: int = ..., ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -246,6 +250,7 @@ def subscriber( max_workers: int = ..., ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -296,6 +301,7 @@ def subscriber( max_workers: int | None = None, ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -351,6 +357,7 @@ def subscriber( max_workers: int | None = None, ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -553,6 +560,9 @@ def subscriber( no_ack: Whether to disable **FastStream** auto acknowledgement logic or not. ack_policy: Acknowledgement policy for the subscriber. no_reply: Whether to disable **FastStream** RPC and Reply To auto responses or not. + exception_handler: Sync or async exception handler called before + the broker's handler. Return True if handled, otherwise False + to delegate to the broker. title: Specification subscriber object title. description: Specification subscriber object description. " "Uses decorated docstring as default. include_in_schema: Whetever to include operation in Specification schema or not. @@ -596,6 +606,7 @@ def subscriber( partitions=partitions, ack_policy=ack_policy, no_reply=no_reply, + exception_handler=exception_handler, config=cast("KafkaBrokerConfig", self.config), # Specification title_=title, diff --git a/faststream/kafka/subscriber/config.py b/faststream/kafka/subscriber/config.py index 6a7616cae9..f6edbaa256 100644 --- a/faststream/kafka/subscriber/config.py +++ b/faststream/kafka/subscriber/config.py @@ -35,6 +35,7 @@ class KafkaSubscriberConfig(SubscriberUsecaseConfig): partitions: Iterable["TopicPartition"] = field(default_factory=list) def __post_init__(self) -> None: + super().__post_init__() self.connection_args["enable_auto_commit"] = self.ack_first @property diff --git a/faststream/kafka/subscriber/factory.py b/faststream/kafka/subscriber/factory.py index bd7f4468cc..66159ef976 100644 --- a/faststream/kafka/subscriber/factory.py +++ b/faststream/kafka/subscriber/factory.py @@ -4,6 +4,7 @@ from faststream._internal.constants import EMPTY from faststream._internal.endpoint.subscriber.call_item import CallsCollection +from faststream._internal.utils import ensure_exception_handler from faststream.exceptions import SetupError from faststream.kafka._compat import validate_client_rack from faststream.kafka.schemas import Topic @@ -21,6 +22,7 @@ if TYPE_CHECKING: from aiokafka.abc import ConsumerRebalanceListener + from faststream._internal.types import ExceptionHandler from faststream.kafka.configs import KafkaBrokerConfig from faststream.kafka.schemas import TopicPartition @@ -40,6 +42,7 @@ def create_subscriber( ack_policy: "AckPolicy", max_workers: int, no_reply: bool, + exception_handler: "ExceptionHandler | None", config: "KafkaBrokerConfig", # Specification args title_: str | None, @@ -71,6 +74,7 @@ def create_subscriber( listener=listener, pattern=pattern, no_reply=no_reply, + exception_handler=ensure_exception_handler(exception_handler), _outer_config=config, _ack_policy=ack_policy, ) diff --git a/tests/brokers/base/exception_handlers.py b/tests/brokers/base/exception_handlers.py new file mode 100644 index 0000000000..0de10d3f50 --- /dev/null +++ b/tests/brokers/base/exception_handlers.py @@ -0,0 +1,1047 @@ +from typing import Any +from unittest.mock import MagicMock, call + +import pytest + +from faststream import Context, FastStream, TestApp +from faststream.exceptions import StopApplication, StopConsume +from faststream.message import AckStatus, StreamMessage +from faststream.middlewares import AckPolicy +from tests.brokers.base.basic import BaseTestcaseConfig + + +@pytest.mark.asyncio() +class ExceptionHandlersTestcase(BaseTestcaseConfig[Any]): + async def test_exception_handlers_are_not_called_for_successful_message( + self, + queue: str, + mock: MagicMock, + mock2: MagicMock, + ) -> None: + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return True + + broker = self.get_broker(exception_handler=exception_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + mock2(msg) + + app = FastStream(broker, exception_handler=exception_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish("hello", queue) + + mock2.assert_called_once_with("hello") + mock.assert_not_called() + + @pytest.mark.parametrize("error_type", (StopConsume, StopApplication)) + async def test_exception_handlers_do_not_suppress_stop_commands( + self, + queue: str, + mock: MagicMock, + error_type: type[BaseException], + ) -> None: + error = error_type() + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return True + + broker = self.get_broker(exception_handler=exception_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=exception_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + with pytest.raises(error_type) as exc_info: + await br.publish(None, queue) + + assert exc_info.value is error + mock.assert_not_called() + + async def test_subscriber_exception_handler_true_suppresses_error( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return True + + broker = self.get_broker() + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + await br.publish(None, queue) + + mock.assert_called_once_with(error) + + async def test_subscriber_exception_handler_receives_context( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def exception_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock(exc, service) + return True + + broker = self.get_broker(apply_types=True) + broker.context.set_global("error_service", expected_service) + + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + await br.publish(None, queue) + + mock.assert_called_once_with(error, expected_service) + + async def test_subscriber_handles_error_without_calling_broker( + self, + queue: str, + mock: MagicMock, + mock2: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def local_handler(exc: BaseException) -> bool: + mock(exc) + return True + + async def broker_handler(exc: BaseException) -> bool: + mock2(exc) + return False + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + await br.publish(None, queue) + + mock.assert_called_once_with(error) + mock2.assert_not_called() + + async def test_broker_handles_error_after_subscriber_declines( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return False + + async def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return True + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + await br.publish(None, queue) + + assert mock.call_args_list == [ + call("subscriber", error), + call("broker", error), + ] + + async def test_broker_exception_handler_true_suppresses_error( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return True + + broker = self.get_broker(exception_handler=exception_handler) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + await br.publish(None, queue) + + mock.assert_called_once_with(error) + + async def test_broker_exception_handler_receives_context( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def exception_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock(exc, service) + return True + + broker = self.get_broker( + apply_types=True, + exception_handler=exception_handler, + ) + broker.context.set_global("error_service", expected_service) + + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + await br.publish(None, queue) + + mock.assert_called_once_with(error, expected_service) + + async def test_subscriber_exception_handler_false_propagates_error( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return False + + broker = self.get_broker() + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await br.publish(None, queue) + + mock.assert_called_once_with(error) + assert exc_info.value is error + + async def test_error_propagates_when_both_handlers_decline( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return False + + def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return False + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await br.publish(None, queue) + + assert mock.call_args_list == [ + call("subscriber", error), + call("broker", error), + ] + assert exc_info.value is error + + async def test_broker_exception_handler_false_propagates_error( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return False + + broker = self.get_broker(exception_handler=exception_handler) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await br.publish(None, queue) + + mock.assert_called_once_with(error) + assert exc_info.value is error + + async def test_broker_exception_handlers_are_independent( + self, + queue: str, + mock: MagicMock, + mock2: MagicMock, + ) -> None: + error1 = RuntimeError("first broker processing failed") + error2 = RuntimeError("second broker processing failed") + + async def exception_handler1(exc: BaseException) -> bool: + mock(exc) + return True + + async def exception_handler2(exc: BaseException) -> bool: + mock2(exc) + return True + + broker1 = self.get_broker(exception_handler=exception_handler1) + broker2 = self.get_broker(exception_handler=exception_handler2) + + args, kwargs = self.get_subscriber_params(queue) + args2, kwargs2 = self.get_subscriber_params(queue + "1") + subscriber1 = broker1.subscriber(*args, **kwargs) + subscriber2 = broker2.subscriber(*args2, **kwargs2) + + @subscriber1 + async def handler1(msg: Any) -> None: + raise error1 + + @subscriber2 + async def handler2(msg: Any) -> None: + raise error2 + + app = FastStream(broker1, broker2) + + async with ( + self.patch_broker(broker1, connect_only=True) as br1, + self.patch_broker(broker2, connect_only=True) as br2, + TestApp(app), + ): + await br1.publish(None, queue) + + mock2.assert_not_called() + + await br2.publish(None, queue + "1") + + mock.assert_called_once_with(error1) + mock2.assert_called_once_with(error2) + + @pytest.mark.parametrize("sync_handler", (False, True)) + async def test_app_exception_handler_true_suppresses_error( + self, + queue: str, + mock: MagicMock, + sync_handler: bool, + ) -> None: + error = RuntimeError("message processing failed") + + def sync_exception_handler(exc: BaseException) -> bool: + mock(exc) + return True + + async def async_exception_handler(exc: BaseException) -> bool: + mock(exc) + return True + + broker = self.get_broker() + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream( + broker, + exception_handler=( + sync_exception_handler if sync_handler else async_exception_handler + ), + ) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + mock.assert_called_once_with(error) + + async def test_app_exception_handler_false_propagates_error( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return False + + broker = self.get_broker() + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=exception_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await br.publish(None, queue) + + mock.assert_called_once_with(error) + assert exc_info.value is error + + async def test_app_handles_error_after_subscriber_and_broker_decline( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return False + + async def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return False + + async def app_handler(exc: BaseException) -> bool: + mock("app", exc) + return True + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=app_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + assert mock.call_args_list == [ + call("subscriber", error), + call("broker", error), + call("app", error), + ] + + async def test_app_handler_is_not_called_after_subscriber_handles_error( + self, + queue: str, + mock: MagicMock, + mock2: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return True + + async def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return True + + async def app_handler(exc: BaseException) -> bool: + mock2(exc) + return True + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=app_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + assert mock.call_args_list == [call("subscriber", error)] + mock2.assert_not_called() + + async def test_app_handler_is_not_called_after_broker_handles_error( + self, + queue: str, + mock: MagicMock, + mock2: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return False + + async def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return True + + async def app_handler(exc: BaseException) -> bool: + mock2(exc) + return True + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=app_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + assert mock.call_args_list == [ + call("subscriber", error), + call("broker", error), + ] + mock2.assert_not_called() + + async def test_error_propagates_when_all_three_handlers_decline( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return False + + def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return False + + def app_handler(exc: BaseException) -> bool: + mock("app", exc) + return False + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=app_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await br.publish(None, queue) + + assert mock.call_args_list == [ + call("subscriber", error), + call("broker", error), + call("app", error), + ] + assert exc_info.value is error + + async def test_app_exception_handler_receives_context( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def exception_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock(exc, service) + return True + + broker = self.get_broker(apply_types=True) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=exception_handler) + app.context.set_global("error_service", expected_service) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + mock.assert_called_once_with(error, expected_service) + + async def test_app_exception_handlers_are_independent( + self, + queue: str, + mock: MagicMock, + mock2: MagicMock, + ) -> None: + error1 = RuntimeError("first app processing failed") + error2 = RuntimeError("second app processing failed") + + async def exception_handler1(exc: BaseException) -> bool: + mock(exc) + return True + + async def exception_handler2(exc: BaseException) -> bool: + mock2(exc) + return True + + broker1, broker2 = self.get_broker(), self.get_broker() + args, kwargs = self.get_subscriber_params(queue) + args2, kwargs2 = self.get_subscriber_params(queue + "1") + subscriber1 = broker1.subscriber(*args, **kwargs) + subscriber2 = broker2.subscriber(*args2, **kwargs2) + + @subscriber1 + async def handler1(msg: Any) -> None: + raise error1 + + @subscriber2 + async def handler2(msg: Any) -> None: + raise error2 + + app1 = FastStream(broker1, exception_handler=exception_handler1) + app2 = FastStream(broker2, exception_handler=exception_handler2) + + async with ( + self.patch_broker(broker1, connect_only=True) as br1, + self.patch_broker(broker2, connect_only=True) as br2, + TestApp(app1), + TestApp(app2), + ): + await br1.publish(None, queue) + mock2.assert_not_called() + + await br2.publish(None, queue + "1") + + mock.assert_called_once_with(error1) + mock2.assert_called_once_with(error2) + + async def test_app_without_exception_handlers_propagates_error( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + broker = self.get_broker() + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + mock(msg) + raise error + + app = FastStream(broker) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await br.publish("hello", queue) + + mock.assert_called_once_with("hello") + assert exc_info.value is error + + async def test_subscriber_exception_handler_receives_dependency_from_app_context( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def exception_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock(exc, service) + return True + + broker = self.get_broker(apply_types=True) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker) + app.context.set_global("error_service", expected_service) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + mock.assert_called_once_with(error, expected_service) + + async def test_broker_exception_handler_receives_dependency_from_app_context( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def exception_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock(exc, service) + return True + + broker = self.get_broker( + apply_types=True, + exception_handler=exception_handler, + ) + args, kwargs = self.get_subscriber_params(queue) + + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker) + app.context.set_global("error_service", expected_service) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + mock.assert_called_once_with(error, expected_service) + + async def test_app_exception_handler_receives_each_broker_context( + self, + queue: str, + mock: MagicMock, + ) -> None: + error1 = RuntimeError("first broker processing failed") + error2 = RuntimeError("second broker processing failed") + + async def exception_handler( + exc: BaseException, + current_broker: Any = Context("broker"), + ) -> bool: + mock(exc, current_broker) + return True + + broker1 = self.get_broker(apply_types=True) + broker2 = self.get_broker(apply_types=True) + args, kwargs = self.get_subscriber_params(queue) + args2, kwargs2 = self.get_subscriber_params(queue + "1") + + @broker1.subscriber(*args, **kwargs) + async def handler1(msg: Any) -> None: + raise error1 + + @broker2.subscriber(*args2, **kwargs2) + async def handler2(msg: Any) -> None: + raise error2 + + app = FastStream(broker1, broker2, exception_handler=exception_handler) + + async with ( + self.patch_broker(broker1, connect_only=True) as br1, + self.patch_broker(broker2, connect_only=True) as br2, + TestApp(app), + ): + await br1.publish(None, queue) + await br2.publish(None, queue + "1") + + assert mock.call_args_list == [ + call(error1, broker1), + call(error2, broker2), + ] + + async def test_exception_handlers_receive_context_after_broker_is_added( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def local_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("subscriber", exc, service) + return False + + async def broker_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("broker", exc, service) + return False + + async def app_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("app", exc, service) + return True + + broker = self.get_broker( + apply_types=True, + exception_handler=broker_handler, + ) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + raise error + + app = FastStream(exception_handler=app_handler) + app.context.set_global("error_service", expected_service) + app.add_broker(broker) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + assert mock.call_args_list == [ + call("subscriber", error, expected_service), + call("broker", error, expected_service), + call("app", error, expected_service), + ] + + async def test_exception_handlers_receive_context_after_router_is_included( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def local_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("subscriber", exc, service) + return False + + async def broker_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("broker", exc, service) + return False + + async def app_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("app", exc, service) + return True + + broker = self.get_broker(apply_types=True, exception_handler=broker_handler) + app = FastStream(broker, exception_handler=app_handler) + app.context.set_global("error_service", expected_service) + router = self.get_router() + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + + @router.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + raise error + + broker.include_router(router) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + assert mock.call_args_list == [ + call("subscriber", error, expected_service), + call("broker", error, expected_service), + call("app", error, expected_service), + ] + + async def test_handled_error_is_suppressed_before_acknowledgement( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + messages: list[StreamMessage[Any]] = [] + + async def exception_handler( + exc: BaseException, + message: StreamMessage[Any] = Context("message"), + ) -> bool: + mock(exc, message.committed) + return True + + broker = self.get_broker(apply_types=True) + args, kwargs = self.get_subscriber_params( + queue, + ack_policy=AckPolicy.NACK_ON_ERROR, + exception_handler=exception_handler, + ) + + @broker.subscriber(*args, **kwargs) + async def handler( + msg: Any, + message: StreamMessage[Any] = Context("message"), + ) -> None: + messages.append(message) + raise error + + async with self.patch_broker(broker) as br: + await br.publish(None, queue) + + assert (mock.call_args_list, [message.committed for message in messages]) == ( + [call(error, None)], + [AckStatus.ACKED], + ) + + async def test_unhandled_error_is_propagated_after_negative_acknowledgement( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + messages: list[StreamMessage[Any]] = [] + + async def exception_handler( + exc: BaseException, + message: StreamMessage[Any] = Context("message"), + ) -> bool: + mock(exc, message.committed) + return False + + broker = self.get_broker(apply_types=True) + args, kwargs = self.get_subscriber_params( + queue, + ack_policy=AckPolicy.NACK_ON_ERROR, + exception_handler=exception_handler, + ) + + @broker.subscriber(*args, **kwargs) + async def handler( + msg: Any, + message: StreamMessage[Any] = Context("message"), + ) -> None: + messages.append(message) + raise error + + async with self.patch_broker(broker) as br: + with pytest.raises(RuntimeError) as exc_info: + await br.publish(None, queue) + + assert ( + mock.call_args_list, + [message.committed for message in messages], + exc_info.value, + ) == ([call(error, None)], [AckStatus.NACKED], error) diff --git a/tests/brokers/kafka/test_exception_handlers.py b/tests/brokers/kafka/test_exception_handlers.py new file mode 100644 index 0000000000..fe7082fa35 --- /dev/null +++ b/tests/brokers/kafka/test_exception_handlers.py @@ -0,0 +1,9 @@ +import pytest + +from tests.brokers.base.exception_handlers import ExceptionHandlersTestcase + +from .basic import KafkaMemoryTestcaseConfig + + +@pytest.mark.kafka() +class TestExceptionHandlers(KafkaMemoryTestcaseConfig, ExceptionHandlersTestcase): ... diff --git a/tests/docs/getting_started/exception_handlers/__init__.py b/tests/docs/getting_started/exception_handlers/__init__.py new file mode 100644 index 0000000000..8b13789179 --- /dev/null +++ b/tests/docs/getting_started/exception_handlers/__init__.py @@ -0,0 +1 @@ + diff --git a/tests/docs/getting_started/exception_handlers/test_handlers.py b/tests/docs/getting_started/exception_handlers/test_handlers.py new file mode 100644 index 0000000000..53f3e83b96 --- /dev/null +++ b/tests/docs/getting_started/exception_handlers/test_handlers.py @@ -0,0 +1,32 @@ +import pytest + +from faststream import TestApp +from tests.marks import require_aiokafka + + +@pytest.mark.kafka() +@pytest.mark.asyncio() +@require_aiokafka +@pytest.mark.parametrize("message", ("subscriber", "broker")) +async def test_kafka_example_suppresses_handled_errors(message: str) -> None: + from docs.docs_src.getting_started.exception_handlers.kafka import app, broker, handle + from faststream.kafka import TestKafkaBroker + + async with TestKafkaBroker(broker, connect_only=True) as br, TestApp(app): + await br.publish(message, "exception-handlers") + + handle.mock.assert_called_once_with(message) + + +@pytest.mark.kafka() +@pytest.mark.asyncio() +@require_aiokafka +async def test_kafka_example_propagates_error_after_app_declines() -> None: + from docs.docs_src.getting_started.exception_handlers.kafka import app, broker, handle + from faststream.kafka import TestKafkaBroker + + async with TestKafkaBroker(broker, connect_only=True) as br, TestApp(app): + with pytest.raises(TypeError, match="application error"): + await br.publish("application", "exception-handlers") + + handle.mock.assert_called_once_with("application")