Function test_event_postprocessor_exception

tests/test_broadcast.py:168–201

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

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

        @event_postprocessor
        async def test_postprocessor():
            raise RuntimeError("test")

        assert test_postprocessor in {
            dependent.call for dependent in message._event_postprocessors
        }

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

        try:
            with app.provider.context({}):
                matcher = on_message()

                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 "RuntimeError: test" in capsys.readouterr().out


@pytest.mark.anyio