Join GitHub today
GitHub is home to over 28 million developers working together to host and review code, manage projects, and build software together.Sign up
Conflicts with pytest-asyncio (sample with redis) #8
Having following test file, things behave differently with different pytest plugins installed:
import aioredis import pytest from aiohttp import web def init(): """Create web.Application instance """ app = web.Application() app.router.add_get('/', hello) app["redis_url"] = "redis://localhost:6379" app.on_startup.append(create_redis_pool) app.on_shutdown.insert(0, close_redis_pool) return app async def hello(request): """url Handler for hello""" return web.Response(text='Hello, world') async def create_redis_pool(app): """Create Redis connection pool""" redis_url = app["redis_url"] pool = await aioredis.create_redis_pool(redis_url, loop=app.loop) app["redis_pool"] = pool async def close_redis_pool(app): """Close Redis connection pool""" pool = app["redis_pool"] pool.close() await pool.wait_closed() # @pytest.fixture # def loop(event_loop): # """Ensure usable event loop for everyone. # # If you comment this fixture out, default pytest-aiohttp one is used # and things start failing (when redis pool is in place). # """ # return event_loop @pytest.fixture def app(loop): """Application instance. """ return init() @pytest.fixture async def cli(test_client, app, loop): """HTTP client for our application. """ return await test_client(app) async def test_hello(cli, loop): resp = await cli.get("/") assert resp.status == 200 text = await resp.text() assert "Hello, world" in text
referenced this issue
Jan 31, 2018
While it's nice that many people can get away with only using the pytest-aiohttp plugin, I don't think this is always possible. I can double check if necessary, but I remember a case where I couldn't just use the aiohttp plugin. I think it was a test that didn't use any aiohttp fixtures and was just a simple async test and got some exception.
Sorry for the worst description ever, but if it would be useful I can try to reproduce what I was seeing.
I think I've encouter same issue with a shorted code using ThreadPoolExecutor without redis.
Following source using
import asyncio import pytest from concurrent.futures import ThreadPoolExecutor def count(): return 0 @pytest.fixture async def my_fixture(event_loop): executor = ThreadPoolExecutor() await event_loop.run_in_executor(executor, count) async def test_f(my_fixture): pass
Following source with
import asyncio import pytest from concurrent.futures import ThreadPoolExecutor def count(): return 0 @pytest.fixture async def my_fixture(loop): executor = ThreadPoolExecutor() await loop.run_in_executor(executor, count) async def test_f(my_fixture): pass