Closed as not planned
Description
Hi, I have a problem with run tests, and I don't know is it bug or I do something wrong.
I use async grpc server and async sqlalchemy in project, and want to write integration test.
I write next fixtures(see code below)
setup_db - migrate on db scope='session' - one time for all tests
session - session db scope='function' - one for every test
grpc_server - start grpc server scope='session', loop_scope='session' - one time for all integration tests
stub - stub for grpc request scope='session', loop_scope='session'
@pytest_asyncio.fixture(scope='session', loop_scope='session', autouse=True)
async def setup_db():
# logger.info("Install test database")
print('Install test database')
async with engine.begin() as conn:
await conn.run_sync(BaseTableORM.metadata.drop_all)
await conn.run_sync(BaseTableORM.metadata.create_all)
@pytest_asyncio.fixture(scope='function', loop_scope='function')
async def session():
async with get_async_session() as session:
yield session
@pytest_asyncio.fixture(scope='session', loop_scope='session', autouse=True)
async def grpc_server():
async def serve() -> None:
server = await init_grpc_server()
await server.start()
await server.wait_for_termination()
server_task = asyncio.create_task(serve())
print(f'Starting grpc server on port {settings.GRPC_SERVICE_PORT}')
yield
server_task.cancel()
@pytest_asyncio.fixture(scope='session', loop_scope='session')
async def stub():
async with grpc.aio.insecure_channel(f'localhost:{settings.GRPC_SERVICE_PORT}') as channel:
yield CoreServiceStub(channel)
I have 2 pack of tests - 2 files
test_func_scope.py
import pytest
from sqlalchemy import text
@pytest.mark.asyncio(loop_scope='function')
async def test_unitest_dummy() -> None:
assert True
@pytest.mark.asyncio(loop_scope='function')
async def test_unitest_session(session) -> None:
result = await session.execute(text('SELECT 1'))
row = result.scalar()
print(f'Database connection is working, result: {row}')
assert True
@pytest.mark.asyncio(loop_scope='function')
async def test_unitest_session2(session) -> None:
result = await session.execute(text('SELECT 1'))
row = result.scalar()
print(f'Database connection is working, result: {row}')
assert True
test_session_scope.py
import pytest
from core.grpc_server.proto.core_service_pb2 import TestRequest
@pytest.mark.asyncio(loop_scope='session')
async def test_stub1(stub) -> None:
await stub.Test(TestRequest(x=1))
assert True
pytest --collect-only
============================================================================= test session starts ==============================================================================
platform darwin -- Python 3.12.0, pytest-8.3.3, pluggy-1.5.0
rootdir: /Users/kras/projects/my/eroc
configfile: pytest.ini
plugins: asyncio-0.24.0, env-1.1.5, anyio-4.6.0, vcr-1.0.2, xdist-3.6.1
asyncio: mode=Mode.AUTO, default_loop_scope=session
collected 4 items
<Dir eroc>
<Package tests>
<Package unitests>
<Module test_func_scope.py>
<Coroutine test_unitest_dummy>
<Coroutine test_unitest_session>
<Coroutine test_unitest_session2>
<Module test_session_scope.py>
<Coroutine test_stub1>
tests with db session fixture works fine!
but my test with grpc server writes me error.
But if I remove test_func_scope.py test_stub1 works fine.
FAILURES ===================================================================================
__________________________________________________________________________________ test_stub1 __________________________________________________________________________________
stub = <core.grpc_server.proto.core_service_pb2_grpc.CoreServiceStub object at 0x1087992b0>
@pytest.mark.asyncio(loop_scope='session')
async def test_stub1(stub) -> None:
> await stub.Test(TestRequest(x=1))
tests/unitests/test_session_scope.py:8:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_AioCall object>
def __await__(self) -> Generator[Any, None, ResponseType]:
"""Wait till the ongoing RPC request finishes."""
try:
> response = yield from self._call_response
E RuntimeError: Task <Task pending name='Task-16' coro=<test_stub1() running at /Users/kras/projects/my/eroc/tests/unitests/test_session_scope.py:8> cb=[_run_until_complete_cb() at /Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/base_events.py:180]> got Future <Task pending name='Task-17' coro=<UnaryUnaryCall._invoke() running at /Users/kras/Library/Caches/pypoetry/virtualenvs/core-orGPXcjT-py3.12/lib/python3.12/site-packages/grpc/aio/_call.py:577>> attached to a different loop
../../../Library/Caches/pypoetry/virtualenvs/core-orGPXcjT-py3.12/lib/python3.12/site-packages/grpc/aio/_call.py:308: RuntimeError
---------------------------------------------------------------------------- Captured log teardown -----------------------------------------------------------------------------
ERROR asyncio:base_events.py:1785 Task was destroyed but it is pending!
task: <Task pending name='Task-20' coro=<AioServer.shutdown() running at src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi:None>>
=============================================================================== warnings summary ===============================================================================
src/core/grpc_server/proto/core_service_pb2.py:0
/Users/kras/projects/my/eroc/src/core/grpc_server/proto/core_service_pb2.py:0: PytestCollectionWarning: cannot collect test class 'TestRequest' because it has a __init__ constructor (from: tests/unitests/test_session_scope.py)
tests/unitests/test_session_scope.py::test_stub1
/Library/Frameworks/Python.framework/Versions/3.12/lib/python3.12/asyncio/base_events.py:689: RuntimeWarning: coroutine 'AioServer.shutdown' was never awaited
self._ready.clear()
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================================================================== short test summary info ============================================================================
FAILED tests/unitests/test_session_scope.py::test_stub1 - RuntimeError: Task <Task pending name='Task-16' coro=<test_stub1() running at /Users/kras/projects/my/eroc/tests/unitests/test_session_scope.py:8> cb=[_run_until_complete_...
and one more remark
If I change
@pytest_asyncio.fixture(scope='session', loop_scope='session')
async def stub():
to
@pytest_asyncio.fixture(scope='function', loop_scope='function')
async def stub():
and
@pytest.mark.asyncio(loop_scope='session')
async def test_stub1(stub) -> None:
to
@pytest.mark.asyncio(loop_scope='function')
async def test_stub1(stub) -> None:
it comes down to the line
await stub.Test(TestRequest(x=1))
and freeze