Testing utilities to detect which handlers were called

[danielknell]

Description

There is a TraceableBus for seeing which messages are passed to a bus, but nothing currently to inspect which handlers were called.

Motivation

Some folk might want to write automated tests to ensure that the correct handler was called for a message.

Possible implementation

Something along the lines of the following, where filename, function, lineno, and timestamp function similarly to the MessageInfo in TraceableBus and reference, handler, name, and request relate to the called handler.

@dataclasses.dataclass(frozen=True)
class HandlerInfo[T]:
    #: handler reference instance
    reference: HandlerReference[T]

    #: handler object
    handler: Handler[T]

    #: module path and class or function name of the handler
    name: str

    #: request object
    request: T

    #: filename where call originated
    filename: str

    #: function where call originated
    function: str

    #: line in file where call originated
    lineno: int

    #: timestamp of call
    timestamp: datetime.datetime

class TraceableHandlerFactory(banshee.HandlerFactory):
    def __init__(self, factory: banshee.HandlerFactory) -> None:
        self.factory = factory
        self._handlers = []

    @property
    def handlers(self):
        return tuple(self._handlers)

    def __call__(self, reference: banshee.HandlerReference[T], /) -> banshee.Handler[T]:
        handler = self.factory(reference)

        @functools.wrap(handler)
        def wrapper(request: T) -> typing.Any:
            self._handlers.append(HandlerInfo(...))

            return handler(request)

        return wrapper

then in a test something like the following can be used:

factory = banshee.SimpleHandlerFactory()

traceable_factory = banshee.TraceableHandlerFactory(factory)

bus = (
   banshee.Builder()
       .with_factory(traceable_factory)

request = FooCommand()

bus.handle(request)

assert any(
    (info.name == "myapp.handlers.handle_foo" and request == request)
    for info in traceable_bus.handlers
)