“Runtime Error: Event loop is closed” 在使用 Pytest 进行测试时发生。

huangapple go评论114阅读模式
英文:

"Runtime Error: Event loop is closed" during testing using Pytest

问题

请帮助我。如何修复它?错误仅在test_logout()中使用cookies时出现。

  1. conftest.py
  2. ...............
  3. @pytest_asyncio.fixture(autouse=True, scope='session')
  4. async def prepare_database():
  5. async with engine_test.begin() as conn:
  6. await conn.run_sync(metadata.create_all)
  7. yield
  8. async with engine_test.begin() as conn:
  9. await conn.run_sync(metadata.drop_all)
  10. @pytest.fixture(scope='session')
  11. def event_loop(request):
  12. loop = asyncio.get_event_loop_policy().new_event_loop()
  13. yield loop
  14. loop.close()
  15. client = TestClient(app)
  16. ...............
  1. test_auth.py
  2. import asyncio
  3. import pytest
  4. import pytest_asyncio
  5. from conftest import client
  6. def test_register():
  7. response = client.post("/auth/register", json={
  8. "email": "user@example.com",
  9. "password": "string",
  10. "is_active": True,
  11. "is_superuser": False,
  12. "is_verified": False
  13. })
  14. assert response.status_code == 201
  15. def test_login():
  16. data = {
  17. "username": "user@example.com",
  18. "password": "string",
  19. }
  20. encoded_data = "&".join([f"{key}={value}" for key, value in data.items()])
  21. headers = {
  22. "Content-Type": "application/x-www-form-urlencoded"
  23. }
  24. response = client.post("/auth/login", data=encoded_data, headers=headers)
  25. assert response.status_code == 204
  26. def test_logout():
  27. cookies = {k: client.cookies.get(k) for k in client.cookies}
  28. print(cookies)
  29. headers = {
  30. "Content-Type": "application/json"
  31. }
  32. response = client.post("/auth/logout", headers=headers, cookies=cookies)
  33. print(response)
  34. 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..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()

  1. conftest.py
  2. ...............
  3. @pytest_asyncio.fixture(autouse=True, scope=&#39;session&#39;)
  4. async def prepare_database():
  5. async with engine_test.begin() as conn:
  6. await conn.run_sync(metadata.create_all)
  7. yield
  8. async with engine_test.begin() as conn:
  9. await conn.run_sync(metadata.drop_all)
  10. @pytest.fixture(scope=&#39;session&#39;)
  11. def event_loop(request):
  12. loop = asyncio.get_event_loop_policy().new_event_loop()
  13. yield loop
  14. loop.close()
  15. client = TestClient(app)
  16. ...............
  1. test_auth.py
  2. import asyncio
  3. import pytest
  4. import pytest_asyncio
  5. from conftest import client
  6. def test_register():
  7. response = client.post(&quot;/auth/register&quot;, json={
  8. &quot;email&quot;: &quot;user@example.com&quot;,
  9. &quot;password&quot;: &quot;string&quot;,
  10. &quot;is_active&quot;: True,
  11. &quot;is_superuser&quot;: False,
  12. &quot;is_verified&quot;: False
  13. })
  14. assert response.status_code == 201
  15. def test_login():
  16. data = {
  17. &quot;username&quot;: &quot;user@example.com&quot;,
  18. &quot;password&quot;: &quot;string&quot;,
  19. }
  20. encoded_data = &quot;&amp;&quot;.join([f&quot;{key}={value}&quot; for key, value in data.items()])
  21. headers = {
  22. &quot;Content-Type&quot;: &quot;application/x-www-form-urlencoded&quot;
  23. }
  24. response = client.post(&quot;/auth/login&quot;, data=encoded_data, headers=headers)
  25. assert response.status_code == 204
  26. def test_logout():
  27. cookies = {k: client.cookies.get(k) for k in client.cookies}
  28. print(cookies)
  29. headers = {
  30. &quot;Content-Type&quot;: &quot;application/json&quot;
  31. }
  32. response = client.post(&quot;/auth/logout&quot;, headers=headers, cookies=cookies)
  33. print(response)
  34. 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

  1. @pytest_asyncio.fixture(scope="session", autouse=False)
  2. async def async_client():
  3. async with AsyncClient(app=app, base_url='http://test') as client:
  4. yield client
  5. await client.close()
  6. @pytest.mark.asyncio
  7. async def test_login(async_client):
  8. await async_client.post(...)
  9. @pytest.mark.asyncio
  10. async def test_logout(async_client):
  11. await async_client.post(...)
英文:

You should create your client in a fixture, and set the scope to &quot;module&quot; or &quot;session&quot; to reuse it across tests.

https://tonybaloney.github.io/posts/async-test-patterns-for-pytest-and-unittest.html

  1. @pytest_asyncio.fixture(scope=&quot;session&quot;, autouse=False)
  2. async def async_client():
  3. async with AsyncClient(app=app, base_url=&#39;http://test&#39;) as client:
  4. yield client
  5. await client.close()
  6. @pytest.mark.asyncio
  7. async def test_login(async_client):
  8. await async_client.post(...)
  9. @pytest.mark.asyncio
  10. async def test_logout(async_client):
  11. await async_client.post(...)

huangapple
  • 本文由 发表于 2023年7月14日 02:20:37
  • 转载请务必保留本文链接:https://go.coder-hub.com/76682245.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定