How to use the anyio.exceptions.ExceptionGroup function in anyio

To help you get started, we’ve selected a few anyio examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github agronholm / anyio / tests / test_taskgroups.py View on Github external
async def test_multi_error_host():
    with pytest.raises(ExceptionGroup) as exc:
        async with create_task_group() as tg:
            await tg.spawn(async_error, 'child', 2)
            await wait_all_tasks_blocked()
            raise Exception('host')

    assert len(exc.value.exceptions) == 2
    assert [str(e) for e in exc.value.exceptions] == ['host', 'child']
    assert exc.match('^2 exceptions were raised in the task group:\n')
    assert exc.match(r'Exception: host\n----')
github agronholm / anyio / tests / test_taskgroups.py View on Github external
async def test_multi_error_children():
    with pytest.raises(ExceptionGroup) as exc:
        async with create_task_group() as tg:
            await tg.spawn(async_error, 'task1')
            await tg.spawn(async_error, 'task2')

    assert len(exc.value.exceptions) == 2
    assert sorted(str(e) for e in exc.value.exceptions) == ['task1', 'task2']
    assert exc.match('^2 exceptions were raised in the task group:\n')
    assert exc.match(r'Exception: task\d\n----')
github agronholm / anyio / tests / test_taskgroups.py View on Github external
"""Test that CancelledErrors are filtered out of nested exception groups."""

    async def fail(name):
        try:
            await anyio.sleep(.1)
        finally:
            raise Exception('%s task failed' % name)

    async def fn():
        async with anyio.create_task_group() as task_group:
            await task_group.spawn(fail, 'parent')
            async with anyio.create_task_group() as task_group2:
                await task_group2.spawn(fail, 'child')
                await anyio.sleep(1)

    with pytest.raises(anyio.exceptions.ExceptionGroup) as exc:
        await fn()

    assert len(exc.value.exceptions) == 2
    assert str(exc.value.exceptions[0]) == 'parent task failed'
    assert str(exc.value.exceptions[1]) == 'child task failed'
github agronholm / anyio / tests / test_networking.py View on Github external
async def test_connrefused(self, target, exception_class, fake_localhost_dns):
        dummy_socket = socket.socket(socket.AF_INET6)
        dummy_socket.bind(('::', 0))
        free_port = dummy_socket.getsockname()[1]
        dummy_socket.close()

        with pytest.raises(OSError) as exc:
            await connect_tcp(target, free_port)

        assert exc.match('All connection attempts failed')
        assert isinstance(exc.value.__cause__, exception_class)
        if exception_class is ExceptionGroup:
            for exc in exc.value.__cause__.exceptions:
                assert isinstance(exc, ConnectionRefusedError)
github agronholm / anyio / tests / test_networking.py View on Github external
            'localhost', ExceptionGroup,
            marks=[pytest.mark.skipif(not socket.has_ipv6, reason='IPv6 is not available')]
        ),
        ('127.0.0.1', ConnectionRefusedError)
    ], ids=['multi', 'single'])
    async def test_connrefused(self, target, exception_class, fake_localhost_dns):
        dummy_socket = socket.socket(socket.AF_INET6)
        dummy_socket.bind(('::', 0))
        free_port = dummy_socket.getsockname()[1]
        dummy_socket.close()

        with pytest.raises(OSError) as exc:
            await connect_tcp(target, free_port)

        assert exc.match('All connection attempts failed')
        assert isinstance(exc.value.__cause__, exception_class)
        if exception_class is ExceptionGroup:
github rmorshea / idom / tests / test_core / test_render.py View on Github external
raise ValueError("this is a bug")

        async def _incoming(self, layout, context, message):
            raise ValueError("this is a bug")

    @idom.element
    async def AnyElement(self):
        return idom.html.div()

    async def send(data):
        pass

    async def recv():
        return {}

    with pytest.raises(ExceptionGroup, match="this is a bug"):
        async with RendererWithBug(idom.Layout(AnyElement())) as renderer:
            await renderer.run(send, recv, None)