fastapi+tortoise-orm测试

概述

fastapi+tortoise的测试比较奇葩，tortoise-orm的测试需要传递event_loop，fastapi的异步测试不能直接访问，就算使用httpx的异步功能也不行（因为不会主动调用startup和shutdown）。
解决方案：
tortsoie-orm的测试解决方案是通过传递event_loop的方式，自己主动激活数据库（当然顺便创建测试数据库等一系列功能）,但是没都要通过event_loop.run_until_complete执行实在是有点难看。

我的解决方案：

1. 自己初始化数据库，redis等异步信息
2. 通过同步或异步方式调用测试（没差）

实现方式：

配置conftest.py

配置conftest.py，配置全局event_loop

import asyncio

import pytest
from tortoise import Tortoise, generate_schema_for_client


@pytest.fixture(scope="session")#配置全局默认线程，注意这个千万不能省
def event_loop():
    return asyncio.get_event_loop()


# 从settings里面粘贴出来并修改
TORTOISE_ORM = {
    "connections": {
        "default": {
            "engine": "tortoise.backends.mysql",
            "credentials": {
                "host": "127.0.0.1",
                "port": 3306,
                "user": "root",
                "password": "mnbvcxz123",
                "database": "fasttmptest",
                "echo": True,
                "maxsize": 10,
            },
        },
    },
    "apps": {
        "fast_tmp": {
            "models": ["fast_tmp.models", "aerich.models", "example.models"],
            "default_connection": "default",
        },
    },
}


@pytest.fixture(scope="session", autouse=True)
async def initialize_tests():
    await Tortoise.init(config=TORTOISE_ORM, _create_db=True)  # 注意，这里的配置是测试数据库
    # 创建数据库
    await generate_schema_for_client(Tortoise.get_connection("default"), safe=True)
    # 尝试删除所有数据库，在所有的数据操作完毕之后回调该方法删除数据库
    yield
    await Tortoise._drop_databases()  # 所有测试完毕销毁测试数据库，如果觉得每次都太慢的话可以重构这个函数

注意：数据库的初始化需要放到fastapi的startup里面去，这样你启动的时候才不会和你原来的数据库冲突。例如：

@main_app.on_event("startup")
async def startup() -> None:
    print("startup")
    await AsyncRedisUtil.init(**settings.REDIS)
    await Tortoise.init(config=settings.TORTOISE_ORM)#把初始化数据库放到这里面
    await rearq.init()

@main_app.on_event("shutdown")
async def shutdown() -> None:
    print("shutdown")
    await AsyncRedisUtil.close()
    await Tortoise.close_connections()
    await rearq.close()

编写测试脚本

当完成上述配置，就可以愉快的测试了

# -*- encoding: utf-8 -*-
"""
@File    : test_fastapi_asyncio.py
@Time    : 2021/1/21 12:35
@Author  : chise
@Email   : [e-mail]
@Software: PyCharm
@info    :
"""
import pytest
from httpx import AsyncClient

from example.main import app

from fast_tmp.models import User


@pytest.mark.asyncio#所有异步都不能少
async def test_root():
    await User.all().count()#愉快的调用tortoise-orm的model进行数据操作
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/fast/auth/login")
    assert response.status_code == 200