Использование HTTPX для тестирования Tornado

Question

Я хотел бы стандартизировать использование HTTPX для тестирования независимо от используемой веб-инфраструктуры Python. Мне удалось заставить его работать с Quart и FastAPI, но у меня проблемы с Tornado, поскольку он не соответствует ASGI и использует конкретную асинхронную реализацию, хотя в настоящее время он основан на asyncio.

Минимальное приложение для тестирования разделено на три части: main.py, conftest.py и test_hello.py.

app / main.py :

from contextlib import contextmanager
from typing import Iterator

from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.web import Application, RequestHandler

from loguru import logger


async def start_resources() -> None:
    '''
    Initialize resources such as async Redis and Database connections
    '''
    logger.info('resources started...')


async def close_resources() -> None:
    '''
    Release resources
    '''
    logger.info('resources closed...')


class HelloHandler(RequestHandler):
    def get(self) -> None:
        self.write({'hello': 'world'})


@contextmanager
def create_app() -> Iterator[Application]:
    IOLoop.current().run_sync(start_resources)
    try:
        app = Application([
            ("/hello", HelloHandler),
        ])
        yield app
    finally:
        IOLoop.current().run_sync(close_resources)


if __name__ == '__main__':
    with create_app() as app:
        http_server = HTTPServer(app)
        http_server.listen(8000)
        logger.info('Listening to port 8000 (use CTRL + C to quit)')
        IOLoop.current().start()

tests / conftest.py :

from typing import Iterator, AsyncIterable

from httpx import AsyncClient
from pytest import fixture
from tornado.platform.asyncio import AsyncIOLoop
from tornado.web import Application

from app.main import create_app  # isort:skip


@fixture
def app(io_loop: AsyncIOLoop) -> Iterator[Application]:
    '''
    Return a Tornado.web.Application object with initialized resources
    '''
    with create_app() as app:
        yield app


@fixture
async def client(app: Application,
                base_url: str) -> AsyncIterable[AsyncClient]:
    async with AsyncClient(base_url=base_url) as _client:
        yield _client

tests / test_hello.py :

from httpx import AsyncClient
from pytest import mark


@mark.gen_test
async def test_hello(client: AsyncClient) -> None:
    resp = await client.get('/hello')
    assert resp.status_code == 200
    assert resp.json() == {'hello': 'world'}

И структура проекта такая:

.
├── app
│   ├── __init__.py
│   └── main.py
├── poetry.lock
├── pyproject.toml
└── tests
    ├── conftest.py
    ├── __init__.py
    └── test_hello.py

И ошибку получаю

$ pytest tests/test_hello.py 
========================================================================== test session starts ==========================================================================
platform linux -- Python 3.6.9, pytest-5.4.3, py-1.8.2, pluggy-0.13.1
rootdir: /tmp/minimal-app
plugins: tornado-0.8.1
collected 1 item                                                                                                                                                        

tests/test_hello.py F                                                                                                                                             [100%]

=============================================================================== FAILURES ================================================================================
______________________________________________________________________________ test_hello _______________________________________________________________________________

client = <async_generator object client at 0x7f78e3de75f8>

    @mark.gen_test
    async def test_hello(client: AsyncClient) -> None:
>       resp = await client.get('/hello')
E       AttributeError: 'async_generator' object has no attribute 'get'

tests/test_hello.py:7: AttributeError
------------------------------------------------------------------------- Captured stderr setup -------------------------------------------------------------------------
2020-06-17 10:21:28.574 | INFO     | app.main:start_resources:15 - resources started...
----------------------------------------------------------------------- Captured stderr teardown ------------------------------------------------------------------------
2020-06-17 10:21:28.595 | INFO     | app.main:close_resources:22 - resources closed...
======================================================================== short test summary info ========================================================================
FAILED tests/test_hello.py::test_hello - AttributeError: 'async_generator' object has no attribute 'get'
=========================================================================== 1 failed in 0.03s ===========================================================================

Answer 1

Я мог бы заставить его работать, заменив pytest-tornado фикстур на пользовательский и добавив alt-pytest-asyncio для поддержки асинхронных тестов. pytest-tornado больше не требуется.

conftest.py :

from typing import AsyncIterable, Iterator

from httpx import AsyncClient
from pytest import fixture
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.platform.asyncio import AsyncIOLoop
from tornado.testing import bind_unused_port
from tornado.web import Application

from app.main import create_app  # isort:skip


@fixture
def io_loop() -> AsyncIOLoop:
    '''
    Copied from https://github.com/eukaryote/pytest-tornasync/blob/master/src/pytest_tornasync/plugin.py#L59-L68
    '''
    loop = IOLoop()
    loop.make_current()
    yield loop
    loop.clear_current()
    loop.close(all_fds=True)


@fixture
def app(io_loop: AsyncIOLoop) -> Iterator[Application]:
    '''
    Return a Tornado.web.Application object with initialized resources
    '''
    with create_app() as app:
        yield app


@fixture
async def client(app: Application) -> AsyncIterable[AsyncClient]:
    '''
    Start a HTTPServer each time
    '''
    http_server = HTTPServer(app)
    port = bind_unused_port()[1]
    http_server.listen(port)
    async with AsyncClient(base_url=f'http://localhost:{port}') as _client:
        yield _client

pyproject.toml :

[tool.poetry.dependencies]
python = "^3.8"
tornado = "^6.0.4"
pytest = "^6.0.1"
httpx = "^0.13.3"
loguru = "^0.5.1"
alt-pytest-asyncio = "^0.5.3"

[tool.poetry.dev-dependencies]

[build-system]
requires = ["poetry>=0.12"]
build-backend = "poetry.masonry.api"