英文:
"Runtime Error: Event loop is closed" during testing using Pytest
问题
请帮助我。如何修复它?错误仅在test_logout()中使用cookies时出现。
conftest.py
...............
@pytest_asyncio.fixture(autouse=True, scope='session')
async def prepare_database():
async with engine_test.begin() as conn:
await conn.run_sync(metadata.create_all)
yield
async with engine_test.begin() as conn:
await conn.run_sync(metadata.drop_all)
@pytest.fixture(scope='session')
def event_loop(request):
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
client = TestClient(app)
...............
test_auth.py
import asyncio
import pytest
import pytest_asyncio
from conftest import client
def test_register():
response = client.post("/auth/register", json={
"email": "user@example.com",
"password": "string",
"is_active": True,
"is_superuser": False,
"is_verified": False
})
assert response.status_code == 201
def test_login():
data = {
"username": "user@example.com",
"password": "string",
}
encoded_data = "&".join([f"{key}={value}" for key, value in data.items()])
headers = {
"Content-Type": "application/x-www-form-urlencoded"
}
response = client.post("/auth/login", data=encoded_data, headers=headers)
assert response.status_code == 204
def test_logout():
cookies = {k: client.cookies.get(k) for k in client.cookies}
print(cookies)
headers = {
"Content-Type": "application/json"
}
response = client.post("/auth/logout", headers=headers, cookies=cookies)
print(response)
assert response.status_code == 201
Pytest:
platform linux -- Python 3.11.3, pytest-7.4.0, pluggy-1.2.0 -- /home/djamal/PycharmProjects/green/venv/bin/python
cachedir: .pytest_cache
rootdir: /home/djamal/PycharmProjects/green/server
plugins: anyio-3.7.1, asyncio-0.21.0
asyncio: mode=Mode.STRICT
collected 3 itemstests/test_auth.py::test_register PASSED [ 33%]
tests/test_auth.py::test_login PASSED [ 66%]
tests/test_auth.py::test_logout FAILED [100%]FAILURES ========================================================================================
test_logout ______________________________________________________________________________________self = Connection<host=localhost,port=6379,db=0>, disable_decoding = False, timeout = None
async def read_response(
self,
disable_decoding: bool = False,
timeout: Optional[float] = None,
*,
disconnect_on_error: bool = True,
):
"""Read the response from a previously sent command"""
read_timeout = timeout if timeout is not None else self.socket_timeout
host_error = self._host_error()
try:
if read_timeout is not None:
async with async_timeout(read_timeout):
response = await self._parser.read_response(
disable_decoding=disable_decoding
)
else:
response = await self._parser.read_response(
disable_decoding=disable_decoding
)../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:782:
_
../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:262: in read_response
response = await self._read_response(disable_decoding=disable_decoding)
../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:270: in _read_response
raw = await self._readline()
../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:344: in _readline
data = await self._stream.readline()
/usr/lib/python3.11/asyncio/streams.py:545: in readline
line = await self.readuntil(sep)
/usr/lib/python3.11/asyncio/streams.py:637: in readuntil
await self._wait_for_data('readuntil')
_self = <StreamReader transport=<SelectorSocketTransport closing fd=18>>, func_name = 'readuntil'
async def _wait_for_data(self, func_name):
"""Wait until feed_data() or feed_eof() is called.If stream was paused, automatically resume it.
"""
StreamReader uses a future to link the protocol feed_data() method
to a read coroutine. Running two read coroutines at the same time
would have an unexpected behaviour. It would not possible to know
which coroutine would get the next data.
if self._waiter is not None:
raise RuntimeError(
f'{func_name}() called while another coroutine is '
f'already waiting for incoming data')assert not self._eof, '_wait_for_data after EOF'
Waiting for data while paused will make deadlock, so prevent it.
This is essential for readexactly(n) for case when n > self._limit.
if self._paused:
self._paused = False
self._transport.resume_reading()self._waiter = self._loop.create_future()
try:
await self._waiter
E RuntimeError: Task <Task pending name='anyio.from_thread.BlockingPortal._call_func' coro=<BlockingPortal._call_func() running at /home/djamal/PycharmProjects/green/venv/lib/python3.11/site-packages/anyio/from_thread.py:217>> cb=[TaskGroup._spawn..task_done() at /home/djamal/PycharmProjects/green/venv/lib/python3.11/site-packages/anyio/_backends/_asyncio.py:661]> got Future attached to a different loop /usr/lib/python3.11/asyncio/streams.py:522: RuntimeError
During handling of the above exception, another exception occurred:
def test_logout():
cookies = {k: client.cookies.get(k) for k in client.cookies}
print(cookies)
headers = {
"Content-Type": "application/json"
}
response = client.post("/auth/logout", headers=headers, cookies=cookies)tests/test_auth.py:38:
_
..........................
_self = <_UnixSelectorEventLoop running=False closed=True debug=False>
def _check_closed(self):
if self._closed:
raise RuntimeError('Event loop is closed')
E RuntimeError: Event loop is closed/usr/lib/python3.11/asyncio/base_events.py:519: RuntimeError
Captured stdout call ----------------------------------------------------------------------------------
{'fastapiusersauth': 'X5LEEnDGUSGXI
英文:
Please help me. How to fix it? The error appears only when using cookies in test_logout()
conftest.py
...............
@pytest_asyncio.fixture(autouse=True, scope='session')
async def prepare_database():
async with engine_test.begin() as conn:
await conn.run_sync(metadata.create_all)
yield
async with engine_test.begin() as conn:
await conn.run_sync(metadata.drop_all)
@pytest.fixture(scope='session')
def event_loop(request):
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
client = TestClient(app)
...............
test_auth.py
import asyncio
import pytest
import pytest_asyncio
from conftest import client
def test_register():
response = client.post("/auth/register", json={
"email": "user@example.com",
"password": "string",
"is_active": True,
"is_superuser": False,
"is_verified": False
})
assert response.status_code == 201
def test_login():
data = {
"username": "user@example.com",
"password": "string",
}
encoded_data = "&".join([f"{key}={value}" for key, value in data.items()])
headers = {
"Content-Type": "application/x-www-form-urlencoded"
}
response = client.post("/auth/login", data=encoded_data, headers=headers)
assert response.status_code == 204
def test_logout():
cookies = {k: client.cookies.get(k) for k in client.cookies}
print(cookies)
headers = {
"Content-Type": "application/json"
}
response = client.post("/auth/logout", headers=headers, cookies=cookies)
print(response)
assert response.status_code == 201
Pytest:
>
> platform linux -- Python 3.11.3, pytest-7.4.0, pluggy-1.2.0 -- /home/djamal/PycharmProjects/green/venv/bin/python
> cachedir: .pytest_cache
> rootdir: /home/djamal/PycharmProjects/green/server
> plugins: anyio-3.7.1, asyncio-0.21.0
> asyncio: mode=Mode.STRICT
> collected 3 items
>
> tests/test_auth.py::test_register PASSED [ 33%]
> tests/test_auth.py::test_login PASSED [ 66%]
> tests/test_auth.py::test_logout FAILED [100%]
>
> FAILURES ========================================================================================
> test_logout ______________________________________________________________________________________
>
> self = Connection<host=localhost,port=6379,db=0>, disable_decoding = False, timeout = None
>
> async def read_response(
> self,
> disable_decoding: bool = False,
> timeout: Optional[float] = None,
> *,
> disconnect_on_error: bool = True,
> ):
> """Read the response from a previously sent command"""
> read_timeout = timeout if timeout is not None else self.socket_timeout
> host_error = self._host_error()
> try:
> if read_timeout is not None:
> async with async_timeout(read_timeout):
> response = await self._parser.read_response(
> disable_decoding=disable_decoding
> )
> else:
> response = await self._parser.read_response(
> disable_decoding=disable_decoding
> )
>
> ../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:782:
> _
> ../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:262: in read_response
> response = await self._read_response(disable_decoding=disable_decoding)
> ../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:270: in _read_response
> raw = await self._readline()
> ../venv/lib/python3.11/site-packages/redis/asyncio/connection.py:344: in _readline
> data = await self._stream.readline()
> /usr/lib/python3.11/asyncio/streams.py:545: in readline
> line = await self.readuntil(sep)
> /usr/lib/python3.11/asyncio/streams.py:637: in readuntil
> await self._wait_for_data('readuntil')
> _
>
> self = <StreamReader transport=<_SelectorSocketTransport closing fd=18>>, func_name = 'readuntil'
>
> async def _wait_for_data(self, func_name):
> """Wait until feed_data() or feed_eof() is called.
>
> If stream was paused, automatically resume it.
> """
> StreamReader uses a future to link the protocol feed_data() method
> to a read coroutine. Running two read coroutines at the same time
> would have an unexpected behaviour. It would not possible to know
> which coroutine would get the next data.
> if self._waiter is not None:
> raise RuntimeError(
> f'{func_name}() called while another coroutine is '
> f'already waiting for incoming data')
>
> assert not self._eof, '_wait_for_data after EOF'
>
> Waiting for data while paused will make deadlock, so prevent it.
> This is essential for readexactly(n) for case when n > self._limit.
> if self._paused:
> self._paused = False
> self._transport.resume_reading()
>
> self._waiter = self._loop.create_future()
> try:
> await self._waiter
> E RuntimeError: Task <Task pending name='anyio.from_thread.BlockingPortal._call_func' coro=<BlockingPortal._call_func() running at /home/djamal/PycharmProjects/green/venv/lib/python3.11/site-packages/anyio/from_thread.py:217> cb=[TaskGroup._spawn.<locals>.task_done() at /home/djamal/PycharmProjects/green/venv/lib/python3.11/site-packages/anyio/_backends/_asyncio.py:661]> got Future <Future pending> attached to a different loop
>
> /usr/lib/python3.11/asyncio/streams.py:522: RuntimeError
>
> During handling of the above exception, another exception occurred:
>
> def test_logout():
> cookies = {k: client.cookies.get(k) for k in client.cookies}
> print(cookies)
> headers = {
> "Content-Type": "application/json"
> }
> response = client.post("/auth/logout", headers=headers, cookies=cookies)
>
> tests/test_auth.py:38:
> _
> ..........................
> _
>
> self = <_UnixSelectorEventLoop running=False closed=True debug=False>
>
> def _check_closed(self):
> if self._closed:
> raise RuntimeError('Event loop is closed')
> E RuntimeError: Event loop is closed
>
> /usr/lib/python3.11/asyncio/base_events.py:519: RuntimeError
> Captured stdout call ----------------------------------------------------------------------------------
> {'fastapiusersauth': 'X5LEEnDGUSGXIhA5gOTQJSyJQ0g7xsKXvx4v2xBFCv8'}
> short test summary info ================================================================================
> FAILED tests/test_auth.py::test_logout - RuntimeError: Event loop is closed
> 1 failed, 2 passed in 1.00s ==============================================================================
>
I've been busy with this for 4 hours, ai gpt didn't help
答案1
得分: 0
你应该在一个 fixture 中创建你的客户端,并将作用范围设置为 "module"
或 "session"
以在测试之间重用它。
https://tonybaloney.github.io/posts/async-test-patterns-for-pytest-and-unittest.html
@pytest_asyncio.fixture(scope="session", autouse=False)
async def async_client():
async with AsyncClient(app=app, base_url='http://test') as client:
yield client
await client.close()
@pytest.mark.asyncio
async def test_login(async_client):
await async_client.post(...)
@pytest.mark.asyncio
async def test_logout(async_client):
await async_client.post(...)
英文:
You should create your client in a fixture, and set the scope to "module"
or "session"
to reuse it across tests.
https://tonybaloney.github.io/posts/async-test-patterns-for-pytest-and-unittest.html
@pytest_asyncio.fixture(scope="session", autouse=False)
async def async_client():
async with AsyncClient(app=app, base_url='http://test') as client:
yield client
await client.close()
@pytest.mark.asyncio
async def test_login(async_client):
await async_client.post(...)
@pytest.mark.asyncio
async def test_logout(async_client):
await async_client.post(...)
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论