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": "[email protected]",
"password": "string",
"is_active": True,
"is_superuser": False,
"is_verified": False
})
assert response.status_code == 201
def test_login():
data = {
"username": "[email protected]",
"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': '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
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(...)