attached to a different loop

[krasoffka]

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

[seifertm]

@krasoffka Can you post the output of pytest --setup-show?
I suspect this is a duplicate of #950.

[krasoffka]

@seifertm

 pytest --setup-show
============================================================================= 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, vcr-1.0.2, anyio-4.6.2.post1, xdist-3.6.1
asyncio: mode=Mode.AUTO, default_loop_scope=session
collected 4 items                                                                                                                                                              

tests/unitests/test_func_scope.py 
SETUP    S event_loop_policy
SETUP    S _session_event_loop (fixtures used: event_loop_policy)
SETUP    S grpc_server
SETUP    S setup_db
        SETUP    F _vcr_marker
        SETUP    F event_loop
        tests/unitests/test_func_scope.py::test_unitest_dummy (fixtures used: _vcr_marker, event_loop, event_loop_policy, grpc_server, request, setup_db).
        TEARDOWN F event_loop
        TEARDOWN F _vcr_marker
        SETUP    F _vcr_marker
        SETUP    F event_loop
        SETUP    F session (fixtures used: event_loop)
        tests/unitests/test_func_scope.py::test_unitest_session (fixtures used: _vcr_marker, event_loop, event_loop_policy, grpc_server, request, session, setup_db).
        TEARDOWN F session
        TEARDOWN F event_loop
        TEARDOWN F _vcr_marker
        SETUP    F _vcr_marker
        SETUP    F event_loop
        SETUP    F session (fixtures used: event_loop)
        tests/unitests/test_func_scope.py::test_unitest_session2 (fixtures used: _vcr_marker, event_loop, event_loop_policy, grpc_server, request, session, setup_db).
        TEARDOWN F session
        TEARDOWN F event_loop
        TEARDOWN F _vcr_marker
tests/unitests/test_session_scope.py 
SETUP    S stub
        SETUP    F _vcr_marker
        tests/unitests/test_session_scope.py::test_stub1 (fixtures used: _session_event_loop, _vcr_marker, event_loop_policy, grpc_server, request, setup_db, stub)F
        TEARDOWN F _vcr_marker
TEARDOWN S stub
TEARDOWN S setup_db
TEARDOWN S grpc_server
TEARDOWN S _session_event_loop
TEARDOWN S event_loop_policy

=================================================================================== FAILURES ===================================================================================
__________________________________________________________________________________ test_stub1 __________________________________________________________________________________

stub = <core.grpc_server.proto.core_service_pb2_grpc.CoreServiceStub object at 0x104732ba0>

    @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_...

[seifertm]

Thanks! I'm pretty certain this is a duplicate of #950.

When the event_loop fixture is torn down, _session_event_loop gets messed up. That's why your tests run when you remove test_func_scope.py.