Gino: How to write unit test (by pytest) when used with Sanic?

[jonahfang]

Please advice, thanks in advance!

[fantix]

个人理解哈，涉及数据库和 Web 的偏业务逻辑的代码，不太适合用狭义上的单元测试来覆盖，可能用一个真实的测试数据库、假数据、加上测试用例所需的请求数据，来做针对不同 URL 的功能性测试更为实际一些。这类测试建议可参考 Sanic 来做：http://sanic.readthedocs.io/en/latest/sanic/testing.html

在这种测试中，除了给不同请求参数、验证响应结果与预期是否一致外，有时也会需要从测试代码里去访问数据库，检查数据变化是否符合预期。这种情况现在 GINO 做起来比较尴尬，因为 test_client 会启动和停止 Sanic 服务器，也就意味着会每次创建和销毁与 db 对象绑定的数据库连接池，而 db 对象通常又尴尬地是一个全局变量，没法再绑定一个测试用的连接池。临时解决方案，可以在测试代码访问数据库时传 bind 参数，显式地把测试连接池传进去。之后 GINO 应该设法支持这种在 Sanic app 的 scope 之外的使用场景。

除此之外，如果确实说需要 mock 掉真实数据库来做单元测试，建议可以考虑将 asyncpg.connection 中的 _execute 函数替换掉，这应该是目前所有数据库访问都要经过的一个底层函数。最后，如果还需要排除 GINO 代码对被测试目标代码的影响，可能就稍微有点麻烦了，毕竟类似于 SQLAlchemy 风格的数据库访问代码句式中的调用比较频繁，很难把所有调用都 mock 掉，当然这样做可能意义也不大。

[fantix]

So generally other than http://sanic.readthedocs.io/en/latest/sanic/testing.html , GINO shall implement something for Sanic so that database can be used without a running Sanic server instance.

[jonahfang]

运行单元测试的时候，会使用一个与开发阶段不同的真实的数据库实例。以我的MacBookPro开发环境而言，开发时使用一个宿主机上的postgresql, 而单元测试的时候，会使用一个postgresql的docker 容器.

我希望是在单元测试方法中，运行测试方法前，准备好数据，然后调用Sanic的url(将Sanic view方法看成black box)，最后验证数据库中的数据是否符合预期。

[fantix]

嗯嗯，很确切。 @lbhsot 可以在 #28 里写一些这种测试。

[jonahfang]

我自己实现了上述涉及到数据库的单元测试，基本思路如下:
1.在conftest中定义下列fixture:

import pytest
import sqlalchemy
from gino import Gino
from .server import app as my_app
from .settings import DB_CONFIG
from apps.demo.models import db as db1
from apps.es_api.models import db as db2

@pytest.fixture
def app():
    yield my_app

@pytest.fixture
def test_cli(loop, app, test_client):
    return loop.run_until_complete(test_client(app))

@pytest.fixture(scope='module')
def db_setup():
    rv = sqlalchemy.create_engine('postgresql://{user}:{password}@{host}:{port}/{database}'.format(**DB_CONFIG))

    db1.create_all(rv)
    db2.create_all(rv)
    yield rv

    db1.drop_all(rv)
    db2.drop_all(rv)
    rv.dispose()

@pytest.fixture
async def pool_setup():
    db = Gino()
    return await db.create_pool('postgresql://{user}:{password}@{host}:{port}/{database}'.format(**DB_CONFIG))

2.单元测试：先在users2表中插入1条记录，完成数据准备，然后调用 Sanic URL后检查返回结果:

@pytest.fixture
async def create_data(pool_setup):
    sql = "INSERT INTO users2 (id,name) VALUES ('1','me')"
    return await exec(None, sql, pool=pool_setup)

async def test_demo_042(db_setup,test_cli,create_data):
    response = await test_cli.get('/d042')
    assert response.status == 200
    rsp = await response.json()
    assert rsp['records'][0]['id']==1
    assert rsp['records'][0]['name']=='me'

请指正。

[jonahfang]

非常感谢作者的辛苦劳动，Gino用起来很舒服，准备用于生产项目。

[fantix]

感谢夸奖！用于生产项目还请谨慎，我们提前为有可能发生的不能及时修正的问题致歉。我们自己也会尽快在生产环境上多尝试尝试。

您的测试代码没有问题，正是我想说 显式地把测试连接池传进去 的 workaround，也是 GINO 自带的测试里的用法。这个 issue 请保留 open，我们尝试下能否再简化一些测试的用法。

[jonahfang]

在 create_data fixture 中如果也能使用models 中的类，而不用写原始的sql， 就更完美了。

[fantix]

啊这里可以用啊：

@pytest.fixture
async def create_data(pool_setup):
    return await User.create(id=1, name='me', bind=pool_setup)

[jonahfang]

@fantix , 完美！谢谢!

[jonahfang]

在单元测试方法中无法使用任何访问数据库的操作，报告错误:

E       RuntimeError: Task <Task pending coro=<test_demo_041() running at /src/apps/demo/tests/test_d04.py:25> cb=[_run_until_complete_cb() at /usr/local/lib/python3.6/asyncio/base_events.py:176]> got Future <Future pending> attached to a different loop

/usr/local/lib/python3.6/site-packages/asyncpg/pool.py:485: RuntimeError
====================================================================== warnings summary ======================================================================
apps/demo/tests/test_d04.py::test_demo_041
  /usr/local/lib/python3.6/asyncio/base_events.py:492: RuntimeWarning: coroutine 'Pool.release.<locals>._release_impl' was never awaited
    self._ready.clear()

-- Docs: http://doc.pytest.org/en/latest/warnings.html

测试程序如下：

import pytest
from gino import Gino
from apps.utils.db import execute,select
from apps.demo.models import User

"""
sh test apps/demo/tests/test_d04.py
"""

@pytest.fixture
async def init_data(pool_setup):
    sql = "DELETE FROM users"
    return await execute(None, sql, pool=pool_setup)  #必须有return或yield: <https://github.com/pytest-dev/pytest-asyncio>

@pytest.fixture
async def prepare_data(pool_setup):
    return await User.create(nickname='me', bind=pool_setup)

async def test_demo_041(db_setup,pool_setup,init_data,test_cli):
    response = await test_cli.get('/demo/d041')
    assert response.status == 200
    rsp = await response.json()
    assert rsp['name']=='daisy中文'

    await User.create(nickname='me1', bind=pool_setup)   #<-------- 此语句出错

#END

[jonahfang]

另外，query 操作时，如何bind到pool?

[fantix]

您可以先试下 from gino.ext.sanic import Gino 来替换 from gino import Gino，然后把 Sanic 的 app 对象传给 db.init_app。

所有 query 方法都支持 bind 参数，比如 User.query.gino.first(bind=pool)

[jonahfang]

改成这样做:

@pytest.fixture
async def prepare_data1(pool_setup):
    sql = "DELETE FROM users"
    yield await execute(None, sql, pool=pool_setup)  #必须有return或yield: <https://github.com/pytest-dev/pytest-asyncio>

    #检查数据库状态
    assert await User.query.gino.first(bind=pool_setup)

async def test_demo_041(db_setup,pool_setup,prepare_data1,test_cli):
    response = await test_cli.get('/demo/d041')
    assert response.status == 200
    rsp = await response.json()
    assert rsp['name']=='daisy中文'

[jonahfang]

在 fixture 中检查执行完 sanic view后的数据库状态

[fantix]

测试多了可能不够用。您也可以尝试在 setup_pool 里，将 app.loop 传到 create_pool 里去，保持 loop 的统一

[jonahfang]

@fantix 不行啊，会报错:

sanic.exceptions.SanicException: Loop can only be retrieved after the app has started running. Not supported with `create_server` function

[fantix]

哦对的，那个需要在 server running 的时候才有。不过看了 Sanic 源码，原来每次 test_cli.get 都会创建新的 asyncio 的 loop 并且设置为全局的 🤣 感觉这个坑有点大了

[fantix]

其实只要能把 run_async=True 塞给 sanic.server.serv 就行

[fantix]

Closing this for now. Please raise new question or discussions in https://github.com/python-gino/gino-sanic.