urllib3 async test setup

conftest.py

from threading import Thread, Event

import pytest
from tornado import web, ioloop

from dummyserver.handlers import TestingApp
from dummyserver.server import DEFAULT_CERTS, run_tornado_app

try:
    import asyncio
except ImportError:
    asyncio = None


@pytest.fixture
def dummy_server_url():
    def run_server_in_thread():
        nonlocal io_loop, server, port

        # The event loop MUST be instantiated in the worker thread, or it will interfere with the
        # tests targeting asyncio
        if asyncio:
            asyncio.set_event_loop(asyncio.new_event_loop())

        io_loop = ioloop.IOLoop.current()
        server, port = run_tornado_app(app, io_loop, DEFAULT_CERTS, scheme, host)
        ready_event.set()
        io_loop.start()

    scheme = "http"
    host = "localhost"
    app = web.Application([(r".*", TestingApp)])
    io_loop = server = port = None
    ready_event = Event()
    server_thread = Thread(target=run_server_in_thread)
    server_thread.start()
    ready_event.wait(5)
    yield "{}://{}:{}".format(scheme, host, port)

    io_loop.add_callback(server.stop)
    io_loop.add_callback(io_loop.stop)
    server_thread.join()

test_poolmanager.py

import pytest

from urllib3 import AsyncPoolManager


@pytest.mark.parametrize(
    "backend",
    [
        pytest.param(
            "trio", id="trio-native", marks=[pytest.mark.anyio(backend="trio")]
        ),
        pytest.param(
            "anyio", id="anyio-trio", marks=[pytest.mark.anyio(backend="trio")]
        ),
        pytest.param(
            "anyio", id="anyio-curio", marks=[pytest.mark.anyio(backend="curio")]
        ),
        pytest.param(
            "anyio", id="anyio-asyncio", marks=[pytest.mark.anyio(backend="asyncio")]
        ),
    ],
)
async def test_redirect(dummy_server_url, backend):
    with AsyncPoolManager(backend=backend) as http:
        r = await http.request(
            "GET",
            "%s/redirect" % dummy_server_url,
            fields={"target": "%s/" % dummy_server_url},
            redirect=False,
        )
        assert r.status == 303

        r = await http.request(
            "GET",
            "%s/redirect" % dummy_server_url,
            fields={"target": "%s/" % dummy_server_url},
        )

        assert r.status == 200
        assert await r.read() == b"Dummy server!"