创建博客主题接口测试报422错误，请求格式不匹配排查

报错详情 #

client = <starlette.testclient.TestClient object at 0x7fe57e1c1bd0>
normal_user_token_headers = {'Authorization': 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ0ZXN0QGV4YW1wbGUuY29tIiwiZXhwIjoxNjgxNzYyODk3fQ.Uhlu3wZmwUSac0ezNcTHwvZntaLWbnvaqVYUIlPwKic'}

    def test_create_topic(client, normal_user_token_headers):
        data = TopicCreate(
            title="451",
            description="here not marketing",
            tags="python",
            image="jpeg.jpg",
        )
        response = client.post("/topics/create-topic/", json=data.dict(), headers=normal_user_token_headers)
>       assert response.status_code == 200
E       assert 422 == 200
E        +  where 422 = <Response [422 Unprocessable Entity]>.status_code

tests/test_routes/test_topics.py:29: AssertionError

相关代码 #

test_topics.py #

import pytest
from pydantic import BaseModel
from datetime import datetime, date


class TopicBase(BaseModel):
    title: str | None = None
    description: str | None = None
    tags: str | None = None
    image: str | None = None
    date_posted: date | None = datetime.now().date().strftime("%m-%d-%Y")


class TopicCreate(TopicBase):
    title: str | None = None
    description: str | None = None
    tags: str | None = None
    image: str | None = None


def test_create_topic(client, normal_user_token_headers):
    data = TopicCreate(
        title="451",
        description="here not marketing",
        tags="python",
        image="jpeg.jpg",
    )
    response = client.post("/topics/create-topic/", json=data.dict(), headers=normal_user_token_headers)
    assert response.status_code == 200

route_topics.py #

router = APIRouter()


@router.post("/create-topic/", response_model=ShowTopic)
def create_topic(
    title: str = Form(...),
    description: str = Form(...),
    image: UploadFile = File(default=None),
    tags: str = Form(...),
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user_from_token),
):
    topics = TopicCreate(title=title, description=description, tags=tags)
    if image:
        topics.image = handle_file_upload(image)
    elif not image:
        topics.image = ""
    return create_new_topic(topic=topics, db=db, owner_id=current_user.id)

schemas.py #

from datetime import date
from datetime import datetime
from typing import Optional

from pydantic import BaseModel


# shared properties
class TopicBase(BaseModel):
    title: str | None = None
    description: str | None = None
    tags: str | None = None
    image: str | None = None
    date_posted: date | None = datetime.now().date()


# this will be used to validate data while creating a Topic
class TopicCreate(TopicBase):
    title: str | None = None
    description: str | None = None
    tags: str | None = None
    image: str | None = None


# this will be used to format the response to not to have id,owner_id etc
class ShowTopic(TopicBase):
    title: str
    description: str
    tags: str
    image: str 
    date_posted: date

    class Config:  # to convert non dict obj to json
        orm_mode = True

conftest.py #

import os
import sys
from typing import Any
from typing import Generator

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# this is to include backend dir in sys.path so that we can import from db,main.py

from db.base import Base
from schemas.topics import TopicCreate
from db.session import get_db
from apis.base import api_router
from core.config import settings
from tests.utils.users import authentication_token_from_email


def start_application():
    app = FastAPI()
    app.include_router(api_router)
    return app


SQLALCHEMY_DATABASE_URL = "sqlite:///./test_db.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# Use connect_args parameter only with sqlite
SessionTesting = sessionmaker(autocommit=False, autoflush=False, bind=engine)


@pytest.fixture(scope="module")
def app() -> Generator[FastAPI, Any, None]:
    """
    Create a fresh database on each test case.
    """
    Base.metadata.create_all(engine)  # Create the tables.
    _app = start_application()
    yield _app
    Base.metadata.drop_all(engine)


@pytest.fixture(scope="module")
def db_session(app: FastAPI) -> Generator[SessionTesting, Any, None]:
    connection = engine.connect()
    transaction = connection.begin()
    session = SessionTesting(bind=connection)
    yield session  # use the session in tests.
    session.close()
    transaction.rollback()
    connection.close()


@pytest.fixture(scope="module")
def client(
    app: FastAPI, db_session: SessionTesting
) -> Generator[TestClient, Any, None]:
    """
    Create a new FastAPI TestClient that uses the `db_session` fixture to override
    the `get_db` dependency that is injected into routes.
    """

    def _get_test_db():
        try:
            yield db_session
        finally:
            pass

    app.dependency_overrides[get_db] = _get_test_db
    with TestClient(app) as client:
        yield client

    
@pytest.fixture(scope="module")
def normal_user_token_headers(client: TestClient, db_session: Session):
    return authentication_token_from_email(
        client=client, email=settings.TEST_USER_EMAIL, db=db_session
        )

排查方向与解决方案 #

核心问题：请求格式不匹配 #

接口create_topic接收的是表单数据（Form/File），但测试代码一开始用json参数发送JSON格式数据，后续改用data参数也未正确处理文件字段，导致FastAPI无法解析请求，返回422错误。

具体解决步骤 #

1. 调整测试代码，发送正确的表单数据

   • 普通字段（title、description、tags）通过data参数传递表单数据
   • 文件字段image可选择不传（用默认None），或通过files参数模拟上传文件

   修改后的测试函数：

   def test_create_topic(client, normal_user_token_headers):
       # 场景1：不传递image字段
       response = client.post(
           "/topics/create-topic/",
           data={
               "title": "451",
               "description": "here not marketing",
               "tags": "python"
           },
           headers=normal_user_token_headers
       )
       assert response.status_code == 200
   
       # 场景2：模拟上传文件
       from io import BytesIO
       response = client.post(
           "/topics/create-topic/",
           data={
               "title": "451",
               "description": "here not marketing",
               "tags": "python"
           },
           files={"image": ("jpeg.jpg", BytesIO(b"fake image content"), "image/jpeg")},
           headers=normal_user_token_headers
       )
       assert response.status_code == 200
   

2. 清理重复的Pydantic定义
   测试文件中自行定义的TopicBase和TopicCreate会与项目schemas.py中的定义冲突，应直接导入项目内的Schema类：

   # test_topics.py 开头修改
   import pytest
   from schemas.topics import TopicCreate
   

3. 统一必填字段约束
   接口中title、description、tags用Form(...)标记为必填，但schemas.py的TopicCreate将其设为可选，需保持一致：

   # schemas.py 中修改TopicCreate
   class TopicCreate(TopicBase):
       title: str
       description: str
       tags: str
       image: str | None = None
   

4. 查看422错误详情
   在测试中打印response.json()可获取具体错误原因，快速定位问题：

   def test_create_topic(client, normal_user_token_headers):
       # 发送请求代码...
       print(response.json())
       assert response.status_code == 200
   

内容的提问来源于stack exchange，提问作者Rauf 44