[Python] RuntimeError: Event loop is closed - unittest, pytest

I have been working on yet another api using fastapi and trying to write test cases for the apis but facing error that event loop is closed

My setup:

so i am using asyncpg driver/library to connect to the main db and for testcases using aiosqlite driver/library to connect with db.sqlite3 file based db

factories.py (kindly ignore this as this is just to generate fake data)

import factory
from factory.enums import CREATE_STRATEGY
from sqlalchemy.orm import Session
from sqlalchemy.ext.asyncio import AsyncSession
from factory.fuzzy import FuzzyChoice
from app.models import User


class SQLAlchemyModelFactory(factory.alchemy.SQLAlchemyModelFactory):

class Meta:
abstract = True
sqlalchemy_session = None
sqlalchemy_session_persistence = "commit"


@classmethod
def set_session(cls, session: Session):
"""Set the session dynamically for all factories inheriting this class"""
cls._meta.sqlalchemy_session = session


@classmethod
async def _save(cls, model_class, session, args, kwargs):
"""Save the model instance using an async session."""
if not isinstance(session, AsyncSession):
raise ValueError("AsyncSQLAlchemyModelFactory requires an AsyncSession")
obj = model_class(*args, **kwargs)
session.add(obj)
await session.commit()
return obj


@classmethod
async def create(cls, **kwargs):
"""Override create to be async."""
return await cls._generate(CREATE_STRATEGY, kwargs)


class UserFactory(SQLAlchemyModelFactory):
class Meta:
model = User


first_name = factory.Faker("first_name")
last_name = factory.Faker("last_name")
email = factory.Faker("email")
password = factory.Faker('text', max_nb_chars=8)
gender = FuzzyChoice(("male", "female", "others"))
mobile = factory.Faker("phone_number")
is_superuser = factory.Faker("boolean")
is_staff = factory.Faker("boolean")
is_active = factory.Faker("boolean")
created_at = factory.Faker("date_time")
updated_at = factory.Faker("date_time")
logged_in_at = factory.Faker("date_time")
logged_out_at = factory.Faker("date_time")


models.py:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import relationship, DeclarativeBase


# TEST_DATABASE_FILENAME = "db.sqlite3"
# SQLALCHEMY_TEST_DATABASE_URL = f"sqlite+aiosqlite:///{TEST_DATABASE_FILENAME}"
# SQLALCHEMY_DATABASE_URL="postgresql+asyncpg://myappassword@localhost:5432/appdb"


engine = create_async_engine(SQLALCHEMY_DATABASE_URL, echo=False, pool_size=10, pool_pre_ping=True)

asyncsession = async_sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession, expire_on_commit=False)

# all the models


app/tests.py (base test class)

import asyncio
import unittest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from fastapi_extensions.models import Base


class AsyncTestCaseHelper:
@staticmethod
async def init(engine):
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)

@staticmethod
async def cleanup(engine):
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await engine.dispose()
if os.path.exists(TEST_DATABASE_FILENAME):
os.remove(TEST_DATABASE_FILENAME)



class AsyncAPITestCase(unittest.IsolatedAsyncioTestCase):
router = None

@classmethod
def setUpClass(cls):
cls.engine = create_async_engine(
SQLALCHEMY_TEST_DATABASE_URL,
connect_args={"check_same_thread": False},
)
cls.asyncsession = async_sessionmaker(
bind=cls.engine, class_=AsyncSession, expire_on_commit=False
)

asyncio.run(AsyncTestCaseHelper.init(cls.engine))


@classmethod
def tearDownClass(cls):
asyncio.run(AsyncTestCaseHelper.cleanup(cls.engine))
cls.engine = None
cls.asyncsession = None


async def asyncSetUp(self):
self.db_session = self.asyncsession()
self.transport = ASGITransport(app=self.router)
self.client = AsyncClient(transport=self.transport, base_url="http://test")


async def asyncTearDown(self):
await self.db_session.rollback()
await self.db_session.close()
await self.client.aclose()
await self.transport.aclose()
await self.engine.dispose()


api/tests.py (the actual file where test cases are written)

import factories
from starlette import status
from api.controller import apirouter
from app.tests import AsyncAPITestCase


class TestToken(AsyncAPITestCase):

router=apirouter

async def asyncSetUp(self):
await super().asyncSetUp()
factories.UserFactory.set_session(session=self.db_session)
factories.PermissionFactory.set_session(session=self.db_session)
factories.ContentTypeFactory.set_session(session=self.db_session)
factories.GroupFactory.set_session(session=self.db_session)

# Create test users
self.correct_user = await factories.UserFactory.create(
email="test@test.com",
password="test@12345"
)
self.incorrect_user = await factories.UserFactory.create(
email="wrong@test.com",
password="wrongpass"
)

await self.db_session.commit()
await self.db_session.refresh(self.correct_user)
await self.db_session.refresh(self.incorrect_user)


async def test_token_generation_with_invalid_credential(self):
url = apirouter.url_path_for("token-signin")
response = await self.client.post(url, json={
"email": self.incorrect_user.email,
"password": self.incorrect_user.password,
})
assert response.status_code == status.HTTP_401_UNAUTHORIZED


async def test_token_generation_with_valid_credential(self):
url = apirouter.url_path_for("token-signin")
response = await self.client.post(url, json={
"email": self.correct_user.email,
"password": self.correct_user.password,
})
assert response.status_code == status.HTTP_200_OK



requirements.txt

fastapi
uvicorn
sqlalchemy
asyncpg
factory_boy
httpx
aiosqlite
pytest
pytest-asyncio
faker


I try to run my testcases: pytest api\tests.py -v

it gives beelow error(whole traceback):

self = <ProactorEventLoop running=False closed=True debug=True>

def _check_closed(self):
if self._closed:
> raise RuntimeError('Event loop is closed')
E RuntimeError: Event loop is closed

C:\Users\Anura\miniconda3\Lib\asyncio\base_events.py:540: RuntimeError

During handling of the above exception, another exception occurred:

self = <api.tests.TestToken testMethod=test_token_generation_with_valid_credential>

async def test_token_generation_with_valid_credential(self):
url = apirouter.url_path_for("token-signin")
> response = await self.client.post(url, json={
"email": self.correct_user.email,
"password": self.correct_user.password,
})

api\tests.py:111:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
venv\Lib\site-packages\httpx\_client.py:1859: in post
return await self.request(
venv\Lib\site-packages\httpx\_client.py:1540: in request
return await self.send(request, auth=auth, follow_redirects=follow_redirects)
venv\Lib\site-packages\httpx\_client.py:1629: in send
response = await self._send_handling_auth(
venv\Lib\site-packages\httpx\_client.py:1657: in _send_handling_auth
response = await self._send_handling_redirects(
venv\Lib\site-packages\httpx\_client.py:1694: in _send_handling_redirects
response = await self._send_single_request(request)
venv\Lib\site-packages\httpx\_client.py:1730: in _send_single_request
response = await transport.handle_async_request(request)
venv\Lib\site-packages\httpx\_transports\asgi.py:170: in handle_async_request
await self.app(scope, receive, send)
venv\Lib\site-packages\starlette\routing.py:714: in __call__
await self.middleware_stack(scope, receive, send)
venv\Lib\site-packages\starlette\routing.py:734: in app
await route.handle(scope, receive, send)
venv\Lib\site-packages\starlette\routing.py:288: in handle
await self.app(scope, receive, send)
venv\Lib\site-packages\starlette\routing.py:76: in app
await wrap_app_handling_exceptions(app, request)(scope, receive, send)
venv\Lib\site-packages\starlette\_exception_handler.py:53: in wrapped_app
raise exc
venv\Lib\site-packages\starlette\_exception_handler.py:42: in wrapped_app
await app(scope, receive, sender)
venv\Lib\site-packages\starlette\routing.py:73: in app
response = await f(request)
venv\Lib\site-packages\fastapi\routing.py:301: in app
raw_response = await run_endpoint_function(
venv\Lib\site-packages\fastapi\routing.py:212: in run_endpoint_function
return await dependant.call(**values)
fastapi_extensions\views.py:148: in dispatch_request
return await super().dispatch_request(request=request)
fastapi_extensions\views.py:76: in dispatch_request
return await method(request)
api\views.py:20: in post
user = await User.fetch(email = data.email, password = data.password, first_only=True)
fastapi_extensions\models.py:77: in fetch
result = await session.execute(select(cls).filter_by(**kwargs))
venv\Lib\site-packages\sqlalchemy\ext\asyncio\session.py:463: in execute
result = await greenlet_spawn(
venv\Lib\site-packages\sqlalchemy\util\_concurrency_py3k.py:201: in greenlet_spawn
result = context.throw(*sys.exc_info())
venv\Lib\site-packages\sqlalchemy\orm\session.py:2365: in execute
return self._execute_internal(
venv\Lib\site-packages\sqlalchemy\orm\session.py:2241: in _execute_internal
conn = self._connection_for_bind(bind)
venv\Lib\site-packages\sqlalchemy\orm\session.py:2110: in _connection_for_bind
return trans._connection_for_bind(engine, execution_options)
<string>:2: in _connection_for_bind
???
venv\Lib\site-packages\sqlalchemy\orm\state_changes.py:139: in _go
ret_value = fn(self, *arg, **kw)
venv\Lib\site-packages\sqlalchemy\orm\session.py:1189: in _connection_for_bind
conn = bind.connect()
venv\Lib\site-packages\sqlalchemy\engine\base.py:3274: in connect
return self._connection_cls(self)
venv\Lib\site-packages\sqlalchemy\engine\base.py:146: in __init__
self._dbapi_connection = engine.raw_connection()
venv\Lib\site-packages\sqlalchemy\engine\base.py:3298: in raw_connection
return self.pool.connect()
venv\Lib\site-packages\sqlalchemy\pool\base.py:449: in connect
return _ConnectionFairy._checkout(self)
venv\Lib\site-packages\sqlalchemy\pool\base.py:1363: in _checkout
with util.safe_reraise():
venv\Lib\site-packages\sqlalchemy\util\langhelpers.py:146: in __exit__
raise exc_value.with_traceback(exc_tb)
venv\Lib\site-packages\sqlalchemy\pool\base.py:1301: in _checkout
result = pool._dialect._do_ping_w_event(
venv\Lib\site-packages\sqlalchemy\engine\default.py:720: in _do_ping_w_event
return self.do_ping(dbapi_connection)
venv\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py:1163: in do_ping
dbapi_connection.ping()
venv\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py:813: in ping
self._handle_exception(error)
venv\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py:794: in _handle_exception
raise error
venv\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py:811: in ping
_ = self.await_(self._async_ping())
venv\Lib\site-packages\sqlalchemy\util\_concurrency_py3k.py:132: in await_only
return current.parent.switch(awaitable) # type: ignore[no-any-return,attr-defined] # noqa: E501
venv\Lib\site-packages\sqlalchemy\util\_concurrency_py3k.py:196: in greenlet_spawn
value = await result
venv\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py:820: in _async_ping
await tr.start()
venv\Lib\site-packages\asyncpg\transaction.py:146: in start
await self._connection.execute(query)
venv\Lib\site-packages\asyncpg\connection.py:349: in execute
result = await self._protocol.query(query, timeout)
asyncpg\\protocol\\protocol.pyx:375: in query
???
asyncpg\\protocol\\protocol.pyx:368: in asyncpg.protocol.protocol.BaseProtocol.query
???
asyncpg\\protocol\\coreproto.pyx:1174: in asyncpg.protocol.protocol.CoreProtocol._simple_query
???
asyncpg\\protocol\\protocol.pyx:967: in asyncpg.protocol.protocol.BaseProtocol._write
???
C:\Users\Anura\miniconda3\Lib\asyncio\proactor_events.py:366: in write
self._loop_writing(data=bytes(data))
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <_ProactorSocketTransport fd=1556 read=<_OverlappedFuture cancelled created at C:\Users\Anura\miniconda3\Lib\asyncio\windows_events.py:506>>, f = None, data = b'Q\x00\x00\x00\x0bBEGIN;\x00'

def _loop_writing(self, f=None, data=None):
try:
if f is not None and self._write_fut is None and self._closing:
# XXX most likely self._force_close() has been called, and
# it has set self._write_fut to None.
return
assert f is self._write_fut
self._write_fut = None
self._pending_write = 0
if f:
f.result()
if data is None:
data = self._buffer
self._buffer = None
if not data:
if self._closing:
self._loop.call_soon(self._call_connection_lost, None)
if self._eof_written:
self._sock.shutdown(socket.SHUT_WR)
# Now that we've reduced the buffer size, tell the
# protocol to resume writing if it was paused. Note that
# we do this last since the callback is called immediately
# and it may add more data to the buffer (even causing the
# protocol to be paused again).
self._maybe_resume_protocol()
else:
> self._write_fut = self._loop._proactor.send(self._sock, data)
E AttributeError: 'NoneType' object has no attribute 'send'


in short error happens here:

async def test_token_generation_with_valid_credential(self):
url = apirouter.url_path_for("token-signin")
> response = await self.client.post(url, json={
"email": self.correct_user.email,
"password": self.correct_user.password,
})

AttributeError: 'NoneType' object has no attribute 'send'
# which is infact due to
def _check_closed(self):
if self._closed:
> raise RuntimeError('Event loop is closed')
E RuntimeError: Event loop is



Note: if i try to run a single test case then it runs but if i try to run 2 or more test cases it fails and show the above error

My attempts:

I tried to switch my test cases based on pure pytest and pytest_asyncio(using fixtures and all) way still having the same issue with that too but i noticed if i use poolclass=NullPool in my both of the connections (asyncpg and aiosqlite) it somehow works but setting asyncpg pool to NullPool won't be the ideal for the production.

Kindly let me know if you need anything from me like my attempts with pytest and pytest_asyncio and all

thanks for your patience and reading this whole thing

Continue reading...