编程测试 FastAPI 应用程序：编写单元测试和集成测试

2024-11-18 22:50:20 +0800 CST views 634

测试 FastAPI 应用程序：编写单元测试和集成测试

测试是软件开发的关键步骤，能够确保应用程序按预期工作，并防止潜在的错误。在 FastAPI 应用程序中，使用单元测试和集成测试不仅能帮助检测问题，还能提升代码的稳定性。本文将带你了解如何为 FastAPI 应用编写单元测试和集成测试，并提供实际的代码演示。

1. FastAPI 应用程序测试简介

FastAPI 提供了非常友好的测试功能，特别是与 pytest 和 httpx 框架的无缝集成，使得测试过程更加轻松。

单元测试：聚焦于独立的功能或方法，不依赖外部资源。
集成测试：测试系统的各个部分是否能够协同工作，通常涉及到 HTTP 请求和数据库交互。

2. 设置测试环境

首先，确保安装了用于测试的依赖库：

pip install pytest httpx

接下来，创建一个简单的 FastAPI 应用进行测试：

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id < 1:
        raise HTTPException(status_code=400, detail="Invalid item ID")
    return {"item_id": item_id, "name": "Item name"}

这个 API 接受一个 item_id 参数，并返回相应的项目数据。

3. 编写单元测试

单元测试关注逻辑的正确性，通常不涉及到 HTTP 请求。让我们为 `read_item` 函数编写测试。

from main import read_item
import pytest
from fastapi import HTTPException

@pytest.mark.asyncio
async def test_read_item_valid():
    response = await read_item(1)
    assert response == {"item_id": 1, "name": "Item name"}

@pytest.mark.asyncio
async def test_read_item_invalid():
    with pytest.raises(HTTPException) as exc_info:
        await read_item(0)
    assert exc_info.value.status_code == 400
    assert exc_info.value.detail == "Invalid item ID"

test_read_item_valid：测试给定有效的 item_id 时是否返回正确结果。
test_read_item_invalid：测试无效的 item_id 是否抛出 HTTPException。

4. 编写集成测试

集成测试模拟用户与 API 进行交互，通常通过 HTTP 请求测试 API 的整体行为。我们将使用 httpx 库发出实际的 HTTP 请求来测试 FastAPI 应用。

import pytest
from httpx import AsyncClient
from main import app

@pytest.mark.asyncio
async def test_read_item_endpoint_valid():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/items/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1, "name": "Item name"}

@pytest.mark.asyncio
async def test_read_item_endpoint_invalid():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/items/0")
    assert response.status_code == 400
    assert response.json() == {"detail": "Invalid item ID"}

test_read_item_endpoint_valid：测试 /items/1 端点的响应是否为 200。
test_read_item_endpoint_invalid：测试 /items/0 端点的响应是否为 400。

5. 运行测试

在终端中运行以下命令来执行测试：

pytest

6. 测试表单处理

FastAPI 支持表单处理，接下来我们添加一个处理登录表单的端点并进行测试。

from fastapi import Form, HTTPException

@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    if username == "test" and password == "secret":
        return {"message": "Login successful"}
    raise HTTPException(status_code=401, detail="Invalid credentials")

测试表单处理：

import pytest
from httpx import AsyncClient
from main import app

@pytest.mark.asyncio
async def test_login_success():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.post("/login/", data={"username": "test", "password": "secret"})
    assert response.status_code == 200
    assert response.json() == {"message": "Login successful"}

@pytest.mark.asyncio
async def test_login_failure():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.post("/login/", data={"username": "wrong", "password": "wrong"})
    assert response.status_code == 401
    assert response.json() == {"detail": "Invalid credentials"}

7. 测试身份验证

接下来我们为 JWT 身份验证编写测试。

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from fastapi import Depends

SECRET_KEY = "secret"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    if form_data.username == "test" and form_data.password == "secret":
        access_token = jwt.encode({"sub": form_data.username}, SECRET_KEY, algorithm=ALGORITHM)
        return {"access_token": access_token, "token_type": "bearer"}
    raise HTTPException(status_code=401, detail="Invalid credentials")

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid credentials")
        return {"username": username}
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid credentials")

测试 JWT 身份验证：

import pytest
from httpx import AsyncClient
from main import app, SECRET_KEY, ALGORITHM
from jose import jwt

@pytest.mark.asyncio
async def test_token_generation():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.post("/token", data={"username": "test", "password": "secret"})
    assert response.status_code == 200
    token = response.json().get("access_token")
    assert token
    decoded_token = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    assert decoded_token["sub"] == "test"

@pytest.mark.asyncio
async def test_protected_route():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        token = jwt.encode({"sub": "test"}, SECRET_KEY, algorithm=ALGORITHM)
        response = await ac.get("/users/me", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 200
    assert response.json() == {"username": "test"}

8. 使用数据库进行测试

我们将模拟数据库操作并测试 FastAPI 中与数据库的交互。

数据库模型与交互：

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

Base.metadata.create_all(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/")
async def create_user(name: str, db: Session = Depends(get_db)):
    db_user = User(name=name)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

测试数据库交互：

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from main import app, Base, get_db

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base.metadata.create_all(bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_create_user():
    response = client.post("/users/", json={"name": "John Doe"})
    assert response.status_code == 200
    assert response.json()["name"] == "John Doe"



def test_read_user():
    response = client.post("/users/", json={"name": "Jane Doe"})
    user_id = response.json()["id"]
    response = client.get(f"/users/{user_id}")
    assert response.status_code == 200
    assert response.json()["name"] == "Jane Doe"

通过以上步骤，你已经掌握了如何为 FastAPI 应用程序编写单元测试和集成测试。通过 pytest、httpx 和模拟技术，你可以确保 FastAPI 应用的全面测试覆盖。持续编写测试能够提高代码质量，降低潜在风险。

复制全文生成海报软件测试 FastAPI Python编程

编程 测试 FastAPI 应用程序：编写单元测试和集成测试