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

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

"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 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()

conftest.py
...............

@pytest_asyncio.fixture(autouse=True, scope=&#39;session&#39;)
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=&#39;session&#39;)
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(&quot;/auth/register&quot;, json={
        &quot;email&quot;: &quot;user@example.com&quot;,
        &quot;password&quot;: &quot;string&quot;,
        &quot;is_active&quot;: True,
        &quot;is_superuser&quot;: False,
        &quot;is_verified&quot;: False
    })
    assert response.status_code == 201


def test_login():
    data = {
        &quot;username&quot;: &quot;user@example.com&quot;,
        &quot;password&quot;: &quot;string&quot;,
    }
    encoded_data = &quot;&amp;&quot;.join([f&quot;{key}={value}&quot; for key, value in data.items()])
    headers = {
        &quot;Content-Type&quot;: &quot;application/x-www-form-urlencoded&quot;
    }
    response = client.post(&quot;/auth/login&quot;, 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 = {
        &quot;Content-Type&quot;: &quot;application/json&quot;
    }
    response = client.post(&quot;/auth/logout&quot;, 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 &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

@pytest_asyncio.fixture(scope=&quot;session&quot;, autouse=False)
async def async_client():
    async with AsyncClient(app=app, base_url=&#39;http://test&#39;) 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(...)

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:

确定