Testing aiohttp application with pytest, aiopg and sqlite

[barr...@gmail.com]

I am trying to add some tests using PyTest and pytest-aiohttp but I am blocked cause I can't find a way to integrate the test with the db.

@pytest.fixture
def cli(loop, test_client):
    app = web.Application(loop=loop)
   
    error_middleware = error_pages({400: handle_400,
                                    404: handle_404,
                                    500: handle_500})
    app.middlewares.append(error_middleware)
    # Jinja2
    aiohttp_jinja2.setup(app, loader=jinja2.FileSystemLoader(os.path.join(C.BASE_DIR, "templates")))


    # secret_key must be 32 url-safe base64-encoded bytes
    fernet_key = fernet.Fernet.generate_key()
    secret_key = base64.urlsafe_b64decode(fernet_key)
    s_setup(app, EncryptedCookieStorage(secret_key))


    # Adding routes
    for route in routes:
        app.router.add_route(route[0], route[1], route[2], name=route[3])


    # Add static route
    app.router.add_static('/static/',
                          path=os.path.join(C.BASE_DIR, "static"),
                          name='static')


    # db connect
    async def create_engine():
        from aiopg.sa import create_engine
        from factory import metadata
        engine = await create_engine(
                user=user,
                database=db,
                host=host,
                password=password,
                loop=loop
                )
        metadata.create_all(engine)
        return engine
   
    app['db'] = create_engine()
    print(dir(app['db']))
    logging.config.dictConfig(C.DEFAULT_LOGGING)
    logger = logging.getLogger('default')
    app['logger'] = logger
    return loop.run_until_complete(test_client(app))

This is the error that I get

AttributeError: 'coroutine' object has no attribute 'acquire'

and is referenced to this line of code

engine = self.request.app['db']
async with engine.acquire() as conn:

Another think is that would be much more easier to sqlite instead of postgreSQL for testing purpose.

Do you think is possible to use 

engine = create_engine('sqlite://')

somehow ?

[Nickolai Novik]

Hi, just add:
app['db'] = loop.run_until_complete(create_engine())



instead of:
app['db'] = create_engine()




Regarding sqlite, aiopg.sa is not full featured SQLAlchemy, but simple layer between query builder and asyncio.
aioodbc can work with sqlite, but there is no SQLAlchemy layer there.

--
You received this message because you are subscribed to the Google Groups "aio-libs" group.
To unsubscribe from this group and stop receiving emails from it, send an email to aio-libs+u...@googlegroups.com.
To post to this group, send email to aio-...@googlegroups.com.
Visit this group at https://groups.google.com/group/aio-libs.
To view this discussion on the web visit https://groups.google.com/d/msgid/aio-libs/ee8b4760-419e-4740-a50a-c6f61812bac1%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[barr...@gmail.com]

Thanks Nickolai!