Function test_event_preprocessor_exception

tests/test_broadcast.py:91–131

(
    app: App, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
)

@pytest.mark.anyio
async def test_event_preprocessor_exception(
    app: App, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
):
    with monkeypatch.context() as m:
        m.setattr(message, "_event_preprocessors", set())

        @event_preprocessor
        async def test_preprocessor():
            raise RuntimeError("test")

        assert test_preprocessor in {
            dependent.call for dependent in message._event_preprocessors
        }

        runned = False

        async def handler():
            nonlocal runned
            runned = True

        handler_id = logger.add(
            sys.stdout,
            level=0,
            diagnose=False,
            filter=default_filter,
            format=default_format,
        )

        try:
            with app.provider.context({}):
                matcher = on_message(handlers=[handler])

                async with app.test_matcher(matcher) as ctx:
                    bot = ctx.create_bot()
                    event = make_fake_event()()
                    ctx.receive_event(bot, event)
        finally:
            logger.remove(handler_id)

        assert not runned, "matcher should not runned"
        assert "RuntimeError: test" in capsys.readouterr().out


@pytest.mark.anyio