考虑这个 mcve：
需求.txt：
fastapi
httpx
motor
pydantic[email]
python-bsonjs
uvicorn==0.24.0
main.py：
import asyncio
import unittest
from typing import Optional
import motor.motor_asyncio
from bson import ObjectId
from fastapi import APIRouter, Body, FastAPI, HTTPException, Request, status
from fastapi.testclient import TestClient
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from pydantic.functional_validators import BeforeValidator
from typing_extensions import Annotated
# -------- Model --------
PyObjectId = Annotated[str, BeforeValidator(str)]
class ItemModel(BaseModel):
id: Optional[PyObjectId] = Field(alias="_id", default=None)
name: str = Field(...)
email: EmailStr = Field(...)
model_config = ConfigDict(
populate_by_name=True,
arbitrary_types_allowed=True,
json_schema_extra={
"example": {"name": "Jane Doe", "email": "[email protected]"}
},
)
# -------- Router --------
mcve_router = APIRouter()
@mcve_router.post(
"",
response_description="Add new item",
response_model=ItemModel,
status_code=status.HTTP_201_CREATED,
response_model_by_alias=False,
)
async def create_item(request: Request, item: ItemModel = Body(...)):
db_collection = request.app.db_collection
new_bar = await db_collection.insert_one(
item.model_dump(by_alias=True, exclude=["id"])
)
created_bar = await db_collection.find_one({"_id": new_bar.inserted_id})
return created_bar
@mcve_router.get(
"/{id}",
response_description="Get a single item",
response_model=ItemModel,
response_model_by_alias=False,
)
async def show_item(request: Request, id: str):
db_collection = request.app.db_collection
if (item := await db_collection.find_one({"_id": ObjectId(id)})) is not None:
return item
raise HTTPException(status_code=404, detail=f"item {id} not found")
if __name__ == "__main__":
app = FastAPI()
app.include_router(mcve_router, tags=["item"], prefix="/item")
app.db_client = motor.motor_asyncio.AsyncIOMotorClient(
"mongodb://127.0.0.1:27017/?readPreference=primary&appname=MongoDB%20Compass&ssl=false"
)
app.db = app.db_client.mcve_db
app.db_collection = app.db.get_collection("bars")
class TestAsync(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self):
self.client = TestClient(app)
async def asyncTearDown(self):
self.client.app.db_client.close()
def run_async_test(self, coro):
loop = asyncio.get_event_loop()
return loop.run_until_complete(coro)
def test_show_item(self):
bar_data = {"name": "John Doe", "email": "[email protected]"}
create_response = self.client.post("/item", json=bar_data)
self.assertEqual(create_response.status_code, 201)
created_item_id = create_response.json().get("id")
self.assertIsNotNone(created_item_id)
response = self.client.get(f"/item/{created_item_id}")
self.assertEqual(response.status_code, 200)
unittest.main()
当我尝试运行它时，我会遇到此崩溃：
(venv) d:\mcve>python mcve.py
E
======================================================================
ERROR: test_show_item (__main__.TestBarRoutesAsync.test_show_item)
----------------------------------------------------------------------
Traceback (most recent call last):
File "D:\software\python\3.11.3-amd64\Lib\unittest\async_case.py", line 90, in _callTestMethod
if self._callMaybeAsync(method) is not None:
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\software\python\3.11.3-amd64\Lib\unittest\async_case.py", line 112, in _callMaybeAsync
return self._asyncioRunner.run(
^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\software\python\3.11.3-amd64\Lib\asyncio\runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\software\python\3.11.3-amd64\Lib\asyncio\base_events.py", line 653, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^
File "d:\mcve\mcve.py", line 87, in test_show_item
response = self.client.get(f"/item/{created_item_id}")
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\starlette\testclient.py", line 502, in get
return super().get(
^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\httpx\_client.py", line 1055, in get
return self.request(
^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\starlette\testclient.py", line 468, in request
return super().request(
^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\httpx\_client.py", line 828, in request
return self.send(request, auth=auth, follow_redirects=follow_redirects)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\httpx\_client.py", line 915, in send
response = self._send_handling_auth(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\httpx\_client.py", line 943, in _send_handling_auth
response = self._send_handling_redirects(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\httpx\_client.py", line 980, in _send_handling_redirects
response = self._send_single_request(request)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\httpx\_client.py", line 1016, in _send_single_request
response = transport.handle_request(request)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\starlette\testclient.py", line 344, in handle_request
raise exc
File "D:\mcve\venv\Lib\site-packages\starlette\testclient.py", line 341, in handle_request
portal.call(self.app, scope, receive, send)
File "D:\mcve\venv\Lib\site-packages\anyio\from_thread.py", line 288, in call
return cast(T_Retval, self.start_task_soon(func, *args).result())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\software\python\3.11.3-amd64\Lib\concurrent\futures\_base.py", line 456, in result
return self.__get_result()
^^^^^^^^^^^^^^^^^^^
File "D:\software\python\3.11.3-amd64\Lib\concurrent\futures\_base.py", line 401, in __get_result
raise self._exception
File "D:\mcve\venv\Lib\site-packages\anyio\from_thread.py", line 217, in _call_func
retval = await retval_or_awaitable
^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\fastapi\applications.py", line 1054, in __call__
await super().__call__(scope, receive, send)
File "D:\mcve\venv\Lib\site-packages\starlette\applications.py", line 116, in __call__
await self.middleware_stack(scope, receive, send)
File "D:\mcve\venv\Lib\site-packages\starlette\middleware\errors.py", line 186, in __call__
raise exc
File "D:\mcve\venv\Lib\site-packages\starlette\middleware\errors.py", line 164, in __call__
await self.app(scope, receive, _send)
File "D:\mcve\venv\Lib\site-packages\starlette\middleware\exceptions.py", line 62, in __call__
await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
File "D:\mcve\venv\Lib\site-packages\starlette\_exception_handler.py", line 55, in wrapped_app
raise exc
File "D:\mcve\venv\Lib\site-packages\starlette\_exception_handler.py", line 44, in wrapped_app
await app(scope, receive, sender)
File "D:\mcve\venv\Lib\site-packages\starlette\routing.py", line 746, in __call__
await route.handle(scope, receive, send)
File "D:\mcve\venv\Lib\site-packages\starlette\routing.py", line 288, in handle
await self.app(scope, receive, send)
File "D:\mcve\venv\Lib\site-packages\starlette\routing.py", line 75, in app
await wrap_app_handling_exceptions(app, request)(scope, receive, send)
File "D:\mcve\venv\Lib\site-packages\starlette\_exception_handler.py", line 55, in wrapped_app
raise exc
File "D:\mcve\venv\Lib\site-packages\starlette\_exception_handler.py", line 44, in wrapped_app
await app(scope, receive, sender)
File "D:\mcve\venv\Lib\site-packages\starlette\routing.py", line 70, in app
response = await func(request)
^^^^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\fastapi\routing.py", line 299, in app
raise e
File "D:\mcve\venv\Lib\site-packages\fastapi\routing.py", line 294, in app
raw_response = await run_endpoint_function(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\fastapi\routing.py", line 191, in run_endpoint_function
return await dependant.call(**values)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "d:\mcve\mcve.py", line 57, in show_item
if (item := await db_collection.find_one({"_id": ObjectId(id)})) is not None:
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\motor\metaprogramming.py", line 75, in method
return framework.run_on_executor(
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\mcve\venv\Lib\site-packages\motor\frameworks\asyncio\__init__.py", line 85, in run_on_executor
return loop.run_in_executor(_EXECUTOR, functools.partial(fn, *args, **kwargs))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\software\python\3.11.3-amd64\Lib\asyncio\base_events.py", line 816, in run_in_executor
self._check_closed()
File "D:\software\python\3.11.3-amd64\Lib\asyncio\base_events.py", line 519, in _check_closed
raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
----------------------------------------------------------------------
Ran 1 test in 0.074s
FAILED (errors=1)
产生崩溃的线路是
response = self.client.get(f"/item/{created_item_id}")顺便说一句，对使用 pytest 根本不感兴趣，这个问题的主要目的是找出问题所在以及如何修复当前的 mcve
提前致谢！
您没有与 /item/{created_item_id} 匹配的端点。然而你拥有的是 /{id} 。因此，您会收到一个隐蔽的 404 notfound 异常，它不会冒泡，但仍然会破坏测试场景，导致事件循环停止。