She*_*ver 15 python pytest prisma pytest-asyncio fastapi
我正在尝试为我的 fastapi 应用程序编写一些测试
\n\n\n我正在使用
\nprisma-client-py数据库。我不知道这是否会改变什么
一切都按预期工作，除了第一个和最后一个之外，它们都因相同的错误而失败：
\nRuntimeError: <asyncio.locks.Event object at 0x7f5696832950 [unset]> is bound to a different event loop\n这是我的conftest.py
import os\nimport asyncio\nimport pytest\nfrom typing import Any, AsyncGenerator, Generator, Iterator\n\nfrom fastapi import FastAPI\nfrom fastapi.testclient import TestClient\nfrom prisma import Prisma, register\n\n\nfrom server.database.base import *\nfrom server.config.exceptions import configure_exception_handlers\nfrom server.config.settings import settings\nfrom server.apis import apis\n\n\ndef start_application() -> FastAPI:\n """\n Return a FastAPI app\n """\n _app = FastAPI(\n title=str(settings.TITLE),\n description=str(settings.DESCRIPTION),\n version=str(settings.VERSION),\n )\n configure_exception_handlers(_app)\n _app.include_router(apis)\n return _app\n\n\nTEST_DB_DSN = "postgresql://postgres:postgres@localhost:5433/postgres"\nprisma = Prisma(datasource={"url": TEST_DB_DSN})\n\n\nasync def initialize_db() -> None:\n """\n Initialize the test database\n """\n print("Initializing")\n print("Creating all tables")\n stream = os.popen(f"dotenv -e .env.test prisma db push --skip-generate")\n output = stream.read()\n print(output)\n\n\nasync def teardown_db(client: Prisma) -> None:\n """\n Teardown the test database\n """\n print("Teardown")\n print("Dropping all tables")\n stream = os.popen(\n f\'dotenv -e .env.test prisma db execute --url "{TEST_DB_DSN}" --file "./server/tests/utils/reset_db.sql" \'\n )\n print("Creating all tables")\n stream = os.popen(f"DB_DSN={TEST_DB_DSN} prisma db push --skip-generate")\n output = stream.read()\n print(output)\n\n\n@pytest.fixture(scope="session")\ndef app() -> Generator[FastAPI, Any, None]:\n """\n Initialize the app\n """\n _app = start_application()\n yield _app\n\n\n@pytest.fixture(scope="module")\ndef event_loop() -> Iterator[asyncio.AbstractEventLoop]:\n """\n Initialize the event loop\n """\n loop = asyncio.get_event_loop_policy().new_event_loop()\n\n yield loop\n loop.close()\n\n\n# Test client\n@pytest.fixture(scope="module")\nasync def client(\n app: FastAPI, event_loop: asyncio.BaseEventLoop\n) -> AsyncGenerator[TestClient, None]:\n """\n Initialize the test client\n """\n await initialize_db()\n register(prisma)\n await prisma.connect()\n with TestClient(app) as c:\n yield c\n await prisma.disconnect()\n await teardown_db(client=prisma)\n下面是我的测试
\nimport asyncio\nfrom fastapi.testclient import TestClient\nfrom jose import jwt\nimport pytest\n\nfrom prisma.models import User\n\nfrom server.config.mail import fm\nfrom server.config.settings import settings\nfrom server.constants.user_types import UserType\nfrom server.helpers.security import create_email_confirmation_token\nfrom server.apis.auth.repositories import AuthRepository\n\nauth_repository = AuthRepository()\n\n\ndef check_mail(data, outbox):\n assert data.get("password") is None\n assert len(outbox) == 1\n assert outbox[0]["subject"] == "Welcome to Mafflle - Verify your email"\n assert outbox[0]["From"] == f"{settings.EMAIL_FROM_NAME} <{settings.EMAIL_FROM}>"\n assert outbox[0]["To"] == data["email"]\n\n\n@pytest.mark.asyncio\nasync def test_signup_user(client: TestClient, event_loop: asyncio.AbstractEventLoop):\n """\n Test the /v1/auth/signup/ endpoint.\n\n This endpoint should return a 201 status code and a\n JSON response with the full user object.\n """\n fm.config.SUPPRESS_SEND = 1\n with fm.record_messages() as outbox:\n payload = {\n "username": "test_user",\n "password": "Password123!",\n "email": "testapp.sheyzi@gmail.com",\n }\n response = client.post("/v1/auth/signup/", json=payload)\n data = response.json()\n\n assert response.status_code == 201\n assert data["username"] == "test_user"\n assert data["email"] == "testapp.sheyzi@gmail.com"\n assert data["user_type"] == "USER"\n assert data["email_confirmed"] == False\n assert data["is_active"] == True\n check_mail(data, outbox)\n\n # confirm email\n confirmation_token = create_email_confirmation_token(data["email"])\n params = {"token": confirmation_token}\n response = client.get("/v1/auth/confirm-email/", params=params)\n assert response.status_code == 200\n\n\n@pytest.mark.asyncio\nasync def test_signup_business(\n client: TestClient, event_loop: asyncio.AbstractEventLoop\n):\n """\n Test the /v1/auth/signup?user_type=business endpoint.\n\n This endpoint should return a 201 status code and a\n JSON response with the full user object.\n """\n fm.config.SUPPRESS_SEND = 1\n with fm.record_messages() as outbox:\n payload = {\n "username": "test_business",\n "password": "Password123!",\n "email": "blogsedap@gmail.com",\n }\n params = {"user_type": "BUSINESS"}\n response = client.post("/v1/auth/signup/", params=params, json=payload)\n data = response.json()\n\n assert response.status_code == 201\n assert data["username"] == "test_business"\n assert data["email"] == "blogsedap@gmail.com"\n assert data["user_type"] == "BUSINESS"\n assert data["email_confirmed"] == False\n assert data["is_active"] == True\n check_mail(data, outbox)\n\n\ndef test_signup_user_with_existing_username(\n client: TestClient, event_loop: asyncio.AbstractEventLoop\n):\n """\n Test the /v1/auth/signup endpoint with an existing username.\n\n This endpoint should return a 400 status code and a\n JSON response with the error message.\n """\n payload = {\n "username": "test_user",\n "password": "Password123!",\n "email": "testapp.sheyzi@gmail.com",\n }\n # Create a user first\n client.post("/v1/auth/signup/", json=payload)\n # Then try to create a user with the same username\n response = client.post("/v1/auth/signup/", json=payload)\n data = response.json()\n assert response.status_code == 400\n assert data["detail"] == "User with this username already exists."\n\n\ndef test_signup_user_with_existing_email(\n client: TestClient, event_loop: asyncio.AbstractEventLoop\n):\n """\n Test the /v1/auth/signup endpoint with an existing email.\n\n This endpoint should return a 400 status code and a\n JSON response with the error message.\n """\n payload = {\n "username": "test_user_2",\n "password": "Password123!",\n "email": "blogsedap@gmail.com",\n }\n # Create a user first\n client.post("/v1/auth/signup/", json=payload)\n # Then try to create a user with the same email\n response = client.post("/v1/auth/signup/", json=payload)\n data = response.json()\n assert response.status_code == 400\n assert data["detail"] == "User with this email already exists."\n\n\ndef test_signup_user_with_invalid_password(\n client: TestClient, event_loop: asyncio.AbstractEventLoop\n):\n """\n Test the /v1/auth/signup endpoint with an invalid password.\n\n This endpoint should return a 400 status code and a\n JSON response with the error message.\n """\n payload = {\n "username": "test_user_3",\n "password": "Password",\n "email": "test_app1@mafflle.com",\n }\n response = client.post("/v1/auth/signup/", json=payload)\n data = response.json()\n assert response.status_code == 400\n assert (\n data["detail"]\n == "Password must be at least 8 characters long and contain at least one"\n " number,one uppercase letter and one special character."\n )\n\n\n@pytest.mark.asyncio\nasync def test_login_user(client: TestClient, event_loop: asyncio.AbstractEventLoop):\n """\n Test the /v1/auth/login endpoint.\n\n This endpoint should return a 200 status code and a\n JSON response with the full user object.\n """\n # Create a user first\n payload = {\n "username": "test_user",\n "password": "Password123!",\n "email": "testapp.sheyzi@gmail.com",\n }\n client.post("/v1/auth/signup/", json=payload)\n # login\n payload = {"identity": "test_user", "password": "Password123!"}\n response = client.post("/v1/auth/login/", json=payload)\n assert response.status_code == 200\n access_token = response.cookies["access_token"]\n refresh_token = response.cookies["refresh_token"]\n assert access_token is not None\n assert refresh_token is not None\n access_token_data = jwt.decode(\n token=access_token, key=settings.AUTH_SECRET, algorithms=["HS256"]\n )\n assert access_token_data["scope"] == "access_token"\n user = await auth_repository.get_user_by_id(user_id=access_token_data["sub"])\n assert user is not None\n assert user.username == "test_user"\n assert user.email == "testapp.sheyzi@gmail.com"\n assert user.user_type == UserType.USER\n assert user.email_confirmed == True\n assert user.is_active == True\n\n # refresh token\n refresh_token_data = jwt.decode(\n token=refresh_token, key=settings.AUTH_SECRET, algorithms=["HS256"]\n )\n assert refresh_token_data["sub"] == str(user.id)\n assert refresh_token_data["scope"] == "refresh_token"\n这是我的错误/失败的详细信息
\n========================================================================== test session starts ==========================================================================\nplatform linux -- Python 3.10.4, pytest-7.1.2, pluggy-1.0.0 -- /home/sheyzi/code/mafflle/mafflle_backend/venv/bin/python3\ncachedir: .pytest_cache\nrootdir: /home/sheyzi/code/mafflle/mafflle_backend, configfile: pyproject.toml\nplugins: asyncio-0.18.3, anyio-3.6.1\nasyncio: mode=auto\ncollected 6 items \n\nserver/tests/test_authentication/test_users.py::test_signup_user FAILED [ 16%]\nserver/tests/test_authentication/test_users.py::test_signup_business PASSED [ 33%]\nserver/tests/test_authentication/test_users.py::test_signup_user_with_existing_username PASSED [ 50%]\nserver/tests/test_authentication/test_users.py::test_signup_user_with_existing_email PASSED [ 66%]\nserver/tests/test_authentication/test_users.py::test_signup_user_with_invalid_password PASSED [ 83%]\nserver/tests/test_authentication/test_users.py::test_login_user FAILED [100%]\n\n=============================================================================== FAILURES ================================================================================\n___________________________________________________________________________ test_signup_user ____________________________________________________________________________\n\nclient = <starlette.testclient.TestClient object at 0x7fb442d59870>, event_loop = <_UnixSelectorEventLoop running=False closed=False debug=False>\n\n @pytest.mark.asyncio\n async def test_signup_user(client: TestClient, event_loop: asyncio.AbstractEventLoop):\n """\n Test the /v1/auth/signup/ endpoint.\n \n This endpoint should return a 201 status code and a\n JSON response with the full user object.\n """\n fm.config.SUPPRESS_SEND = 1\n with fm.record_messages() as outbox:\n payload = {\n "username": "test_user",\n "password": "Password123!",\n "email": "testapp.sheyzi@gmail.com",\n }\n> response = client.post("/v1/auth/signup/", json=payload)\n\nserver/tests/test_authentication/test_users.py:40: \n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _\nvenv/lib/python3.10/site-packages/requests/sessions.py:635: in post\n return self.request("POST", url, data=data, json=json, **kwargs)\nvenv/lib/python3.10/site-packages/starlette/testclient.py:468: in request\n return super().request(\nvenv/lib/python3.10/site-packages/requests/sessions.py:587: in request\n resp = self.send(prep, **send_kwargs)\nvenv/lib/python3.10/site-packages/requests/sessions.py:701: in send\n r = adapter.send(request, **kwargs)\nvenv/lib/python3.10/site-packages/starlette/testclient.py:266: in send\n raise exc\nvenv/lib/python3.10/site-packages/starlette/testclient.py:263: in send\n portal.call(self.app, scope, receive, send)\nvenv/lib/python3.10/site-packages/anyio/from_thread.py:283: in call\n return cast(T_Retval, self.start_task_soon(func, *args).result())\n/usr/lib/python3.10/concurrent/futures/_base.py:446: in result\n return self.__get_result()\n/usr/lib/python3.10/concurrent/futures/_base.py:391: in __get_result\n raise self._exception\nvenv/lib/python3.10/site-packages/anyio/from_thread.py:219: in _call_func\n retval = await retval\nvenv/lib/python3.10/site-packages/fastapi/applications.py:261: in __call__\n await super().__call__(scope, receive, send)\nvenv/lib/python3.10/site-packages/starlette/applications.py:112: in __call__\n await self.middleware_stack(scope, receive, send)\nvenv/lib/python3.10/site-packages/starlette/middleware/errors.py:181: in __call__\n raise exc\nvenv/lib/python3.10/site-packages/starlette/middleware/errors.py:159: in __call__\n await self.app(scope, receive, _send)\nvenv/lib/python3.10/site-packages/starlette/exceptions.py:82: in __call__\n raise exc\nvenv/lib/python3.10/site-packages/starlette/exceptions.py:71: in __call__\n await self.app(scope, receive, sender)\nvenv/lib/python3.10/site-packages/fastapi/middleware/asyncexitstack.py:21: in __call__\n raise e\nvenv/lib/python3.10/site-packages/fastapi/middleware/asyncexitstack.py:18: in __call__\n await self.app(scope, receive, send)\nvenv/lib/python3.10/site-packages/starlette/routing.py:656: in __call__\n await route.handle(scope, receive, send)\nvenv/lib/python3.10/site-packages/starlette/routing.py:259: in handle\n await self.app(scope, receive, send)\nvenv/lib/python3.10/site-packages/starlette/routing.py:61: in app\n response = await func(request)\nvenv/lib/python3.10/site-packages/fastapi/routing.py:227: in app\n raw_response = await run_endpoint_function(\nvenv/lib/python3.10/site-packages/fastapi/routing.py:160: in run_endpoint_function\n return await dependant.call(**values)\nserver/apis/auth/router.py:30: in signup\n return await self.auth_service.signup(\nserver/apis/auth/services.py:82: in signup\n if await self.auth_repository.get_user_by_username_or_email(user.username):\nserver/apis/auth/repositories.py:66: in get_user_by_username_or_email\n user = await User.prisma().find_first(\nvenv/lib/python3.10/site-packages/prisma/actions.py:1389: in find_first\n resp = await self._client._execute(\nvenv/lib/python3.10/site-packages/prisma/client.py:353: in _execute\n return await self._engine.query(builder.build())\nvenv/lib/python3.10/site-packages/prisma/engine/query.py:185: in query\n return await self.request(\'POST\', \'/\', content=content)\nvenv/lib/python3.10/site-packages/prisma/engine/http.py:96: in request\n resp = await self.session.request(method, url, **kwargs)\nvenv/lib/python3.10/site-packages/prisma/_async_http.py:28: in request\n return Response(await self.session.request(method, url, **kwargs))\nvenv/lib/python3.10/site-packages/httpx/_client.py:1506: in request\n return await self.send(request, auth=auth, follow_redirects=follow_redirects)\nvenv/lib/python3.10/site-packages/httpx/_client.py:1593: in send\n response = await self._send_handling_auth(\nvenv/lib/python3.10/site-packages/httpx/_client.py:1621: in _send_handling_auth\n response = await self._send_handling_redirects(\nvenv/lib/python3.10/site-packages/httpx/_client.py:1658: in _send_handling_redirects\n response = await self._send_single_request(request)\nvenv/lib/python3.10/site-packages/httpx/_client.py:1695: in _send_single_request\n response = await transport.handle_async_request(request)\nvenv/lib/python3.10/site-packages/httpx/_transports/default.py:353: in handle_async_request\n resp = await self._pool.handle_async_request(req)\nvenv/lib/python3.10/site-packages/httpcore/_async/connection_pool.py:253: in handle_async_request\n raise exc\nvenv/lib/python3.10/site-packages/httpcore/_async/connection_pool.py:237: in handle_async_request\n response = await connection.handle_async_request(request)\nvenv/lib/python3.10/site-packages/httpcore/_async/connection.py:90: in handle_async_request\n return await self._connection.handle_async_request(request)\nvenv/lib/python3.10/site-packages/httpcore/_async/http11.py:102: in handle_async_request\n raise exc\nvenv/lib/python3.10/site-packages/httpcore/_async/http11.py:81: in handle_async_request\n ) = await self._receive_response_headers(**kwargs)\nvenv/lib/python3.10/site-packages/httpcore/_async/http11.py:143: in _receive_response_headers\n event = await self._receive_event(timeout=timeout)\nvenv/lib/python3.10/site-packages/httpcore/_async/http11.py:172: in _receive_event\n data = await self._network_stream.read(\nvenv/lib/python3.10/site-packages/httpcore/backends/asyncio.py:31: in read\n return await self._stream.receive(max_bytes=max_bytes)\nvenv/lib/python3.10/site-packages/anyio/_backends/_asyncio.py:1265: in receive\n await self._protocol.read_event.wait()\n/usr/lib/python3.10/asyncio/locks.py:211: in wait\n fut = self._get_loop().create_future()\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _\n\nself = <asyncio.locks.Event object at 0x7fb442dda980 [set]>\n\n def _get_loop(self):\n loop = events._get_running_loop()\n \n if self._loop is None:\n with _global_lock:\n if self._loop is None:\n self._loop = loop\n if loop is not self._loop:\n> raise RuntimeError(f\'{self!r} is bound to a different event loop\')\nE RuntimeError: <asyncio.locks.Event object at 0x7fb442dda980 [unset]> is bound to a different event loop\n\n/usr/lib/python3.10/asyncio/mixins.py:30: RuntimeError\n------------------------------------------------------------------------- Captured stdout setup -------------------------------------------------------------------------\nInitializing\nCreating all tables\nEnvironment variables loaded from .env\nPrisma schema loaded from prisma/schema.prisma\nDatasource "db": PostgreSQL database "postgres", schema "public" at "localhost:5433"\n\n Your database is now in sync with your schema. Done in 1.89s\n\n\n____________________________________________________________________________ test_login
和event_loop夹具app具有不同的范围（分别为"module"和"session"。FastAPI 应用程序可以使用不同的循环创建。我建议使用夹具"session"的范围event_loop并在测试模块顶部声明它。
此外，我不确定当同步装置需要循环引用时使用协程的行为是什么。如果它仍然无法在"session"示波器上工作，请尝试使用同步夹具：
# declare this before any other fixtures with the "session" scope
# that may reference the event loop
@pytest.fixture(scope="session")
def event_loop():
return asyncio.get_event_loop()