使用 HTTPX 进行 Tornado 测试

Question

无论使用何种 Python Web 框架，我都希望标准化 HTTPX 的使用以进行测试。我设法让它与 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

测试/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"