Модульное тестирование FastAPI: AttributeError: объект «NoneType» не имеет атрибута «отправить». ⇐ Python

[Anonymous]

Я пытаюсь протестировать свое приложение FastAPI. Я использую настоящую тестовую базу данных Postgres, поэтому мне действительно хочется читать/записывать данные из/в базу данных вместо имитации.
Первый тест работает нормально. Однако второй тест всегда выдает мне эту ошибку:
FAILED tests/users/test_signup.py::test_signup_successful2 - AttributeError: 'NoneType' object has no attribute 'send'

Первый и второй тест в основном идентичны, чтобы гарантировать, что это не ошибка логики теста. Я предполагаю, что это как-то связано с клиентом, который я использую. Раньше я сталкивался с трудностями, так как не мог заставить работать FastAPI TestClient. Поэтому я перешел на AsyncClient, который по какой-то причине работает только для одного теста, и я не знаю почему.
Вот как сейчас выглядит мой код:
# tests/test_signup.py
import pytest
import pytest_asyncio
from httpx import AsyncClient, get
from httpx._transports.asgi import ASGITransport
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.sql import text
from app import app
from database.session import get_session

@pytest_asyncio.fixture
async def client():
"""Fixture for creating a new test client."""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
yield client

@pytest_asyncio.fixture
async def session():
"""Fixture for creating a new test session."""
session: AsyncSession = await get_session().__anext__() # Get the session
try:
yield session
finally:
await session.close() # Ensure the session is properly closed

@pytest.mark.asyncio
async def test_signup_successful(client, session):
"""Test user signup with valid data"""
# Use ASGITransport explicitly
# transport = ASGITransport(app=app)
# async with AsyncClient(transport=transport, base_url="http://test") as client:
# Define the request payload
payload = {
"first_name": "John",
"last_name": "Doe",
"username": "johndoe",
"email": "testuser@example.com",
"password": "strongpassword123"
}
# Perform POST request
response = await client.post("/user/signup", json=payload)

# Assertions
assert response.status_code == 201
data = response.json()
assert data["email"] == payload["email"]
assert "uuid" in data
assert data["role"] == "user"

# Verify the user exists in the database
statement = text(f"SELECT email FROM users WHERE email = '{payload['email']}'")
result = await session.exec(statement)
user = result.scalar()
assert user is not None
await session.close()

@pytest.mark.asyncio
async def test_signup_successful2(client, session):
"""Test user signup with valid data"""
# Use ASGITransport explicitly
# transport = ASGITransport(app=app)
# async with AsyncClient(transport=transport, base_url="http://test") as client:
# Define the request payload
payload = {
"first_name": "John",
"last_name": "Doe2",
"username": "johndoe2",
"email": "testuser2@example.com",
"password": "strongpassword123"
}
# Perform POST request
response = await client.post("/user/signup", json=payload)

# Assertions
assert response.status_code == 201
data = response.json()
assert data["email"] == payload["email"]
assert "uuid" in data
assert data["role"] == "user"

# Verify the user exists in the database
statement = text(f"SELECT email FROM users WHERE email = '{payload['email']}'")
result = await session.exec(statement)
user = result.scalar()
assert user is not None
await session.close()

Вот для справки вся трассировка:
================================================================================================== test session starts ===================================================================================================
platform win32 -- Python 3.11.10, pytest-8.3.3, pluggy-1.5.0
rootdir: C:\Users\myuser\Documents\visual-studio-code\my-project\backend-v2
configfile: pytest.ini
plugins: anyio-4.6.2.post1, asyncio-0.24.0
asyncio: mode=Mode.STRICT, default_loop_scope=function
collected 3 items

tests\test_app.py .
tests\users\test_signup.py .F

======================================================================================================== FAILURES ========================================================================================================
________________________________________________________________________________________________ test_signup_successful2 _________________________________________________________________________________________________

client = , session =

@pytest.mark.asyncio
async def test_signup_successful2(client, session):
"""Test user signup with valid data"""
# Use ASGITransport explicitly
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
# Define the request payload
payload = {
"first_name": "John",
"last_name": "Doe2",
"username": "johndoe2",
"email": "testuser2@example.com",
"password": "strongpassword123"
}
# Perform POST request
> response = await client.post("/user/signup", json=payload)

tests\users\test_signup.py:77:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\httpx\_client.py in post
return await self.request(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\httpx\_client.py in request
return await self.send(request, auth=auth, follow_redirects=follow_redirects)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\httpx\_client.py in send
response = await self._send_handling_auth(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\httpx\_client.py in _send_handling_auth
response = await self._send_handling_redirects(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\httpx\_client.py in _send_handling_redirects
response = await self._send_single_request(request)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\httpx\_client.py in _send_single_request
response = await transport.handle_async_request(request)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\httpx\_transports\asgi.py:157: in handle_async_request
await self.app(scope, receive, send)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\fastapi\applications.py in __call__
await super().__call__(scope, receive, send)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\applications.py:113: in __call__
await self.middleware_stack(scope, receive, send)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\middleware\errors.py:187: in __call__
raise exc
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\middleware\errors.py:165: in __call__
await self.app(scope, receive, _send)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\middleware\base.py:185: in __call__
with collapse_excgroups():
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\contextlib.py:158: in __exit__
self.gen.throw(typ, value, traceback)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\_utils.py:83: in collapse_excgroups
raise exc
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\middleware\base.py:187: in __call__
response = await self.dispatch_func(request, call_next)
middleware.py:27: in execution_timer
response = await call_next(request)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\middleware\base.py:163: in call_next
raise app_exc
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\middleware\base.py:149: in coro
await self.app(scope, receive_or_disconnect, send_no_error)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\middleware\trustedhost.py:36: in __call__
await self.app(scope, receive, send)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\middleware\cors.py:85: in __call__
await self.app(scope, receive, send)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\middleware\exceptions.py:62: in __call__
await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\_exception_handler.py:62: in wrapped_app
raise exc
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\_exception_handler.py:51: in wrapped_app
await app(scope, receive, sender)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\routing.py:715: in __call__
await self.middleware_stack(scope, receive, send)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\routing.py:735: in app
await route.handle(scope, receive, send)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\routing.py:288: in handle
await self.app(scope, receive, send)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\routing.py:76: in app
await wrap_app_handling_exceptions(app, request)(scope, receive, send)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\_exception_handler.py:62: in wrapped_app
raise exc
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\_exception_handler.py:51: in wrapped_app
await app(scope, receive, sender)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\starlette\routing.py:73: in app
response = await f(request)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\fastapi\routing.py:301: in app
raw_response = await run_endpoint_function(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\fastapi\routing.py:212: in run_endpoint_function
return await dependant.call(**values)
api\users\routes.py:36: in create_user_Account
user_exists = await user_service.user_exists(email, session)
api\users\service.py:17: in user_exists
user = await self.get_user_by_email(email, session)
api\users\service.py:12: in get_user_by_email
result = await session.exec(statement)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlmodel\ext\asyncio\session.py:81: in exec
result = await greenlet_spawn(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\util\_concurrency_py3k.py:201: in greenlet_spawn
result = context.throw(*sys.exc_info())
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlmodel\orm\session.py:66: in exec
results = super().execute(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\orm\session.py in execute
return self._execute_internal(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\orm\session.py in _execute_internal
result: Result[Any] = compile_state_cls.orm_execute_statement(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\orm\context.py:305: in orm_execute_statement
result = conn.execute(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\engine\base.py in execute
return meth(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\sql\elements.py:515: in _execute_on_connection
return connection._execute_clauseelement(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\engine\base.py in _execute_clauseelement
ret = self._execute_context(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\engine\base.py in _execute_context
return self._exec_single_context(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\engine\base.py in _exec_single_context
self._handle_dbapi_exception(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\engine\base.py in _handle_dbapi_exception
raise exc_info[1].with_traceback(exc_info[2])
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\engine\base.py in _exec_single_context
self.dialect.do_execute(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\engine\default.py:941: in do_execute
cursor.execute(statement, parameters)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py:568: in execute
self._adapt_connection.await_(
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\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
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\util\_concurrency_py3k.py:196: in greenlet_spawn
value = await result
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py:504: in _prepare_and_execute
await adapt_connection._start_transaction()
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py:833: in _start_transaction
self._handle_exception(error)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py:782: in _handle_exception
raise error
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\sqlalchemy\dialects\postgresql\asyncpg.py:831: in _start_transaction
await self._transaction.start()
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\asyncpg\transaction.py:146: in start
await self._connection.execute(query)
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\site-packages\asyncpg\connection.py:350: in execute
result = await self._protocol.query(query, timeout)
asyncpg\protocol\protocol.pyx:374: in query
???
asyncpg\protocol\protocol.pyx:367: in asyncpg.protocol.protocol.BaseProtocol.query
???
asyncpg\protocol\coreproto.pyx in asyncpg.protocol.protocol.CoreProtocol._simple_query
???
asyncpg\protocol\protocol.pyx:966: in asyncpg.protocol.protocol.BaseProtocol._write
???
..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\asyncio\proactor_events.py:365: in write
self._loop_writing(data=bytes(data))
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = , f = None, data = b'Q\x00\x00\x00*BEGIN ISOLATION LEVEL READ COMMITTED;\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'

..\..\..\..\AppData\Local\miniconda3\envs\aidav2\Lib\asyncio\proactor_events.py:401: AttributeError
================================================================================================ short test summary info =================================================================================================
FAILED tests/users/test_signup.py::test_signup_successful2 - AttributeError: 'NoneType' object has no attribute 'send'
============================================================================================== 1 failed, 2 passed in 1.85s ===============================================================================================


Изменить:
Для справки здесь также есть мой код базы данных/session.py:
# database/session.py
from sqlmodel import SQLModel, create_engine
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlalchemy.orm import sessionmaker
from config import config

engine = AsyncEngine(create_engine(url=config.SQLALCHEMY_DATABASE_URI, echo=config.DB_ECHO))

async def init_db():
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)

async def get_session() -> AsyncSession: # type: ignore
Session = sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False
)
async with Session() as session:
yield session



Подробнее здесь: https://stackoverflow.com/questions/791 ... ibute-send