如何在FastAPI Strawberry中使用独立测试数据库测试GraphQL端点?
解决方案
1. 使用Alembic为测试数据库执行迁移
Alembic支持通过配置切换目标数据库,你可以通过以下方式实现测试库的迁移:
方式一:通过命令行参数指定测试库URL
在测试前调用Alembic命令,传入测试数据库的URL:
import pytest from alembic.config import Config from alembic import command @pytest.fixture(scope="session") def test_db_migrations(): # 加载Alembic配置 alembic_cfg = Config("alembic.ini") # 替换配置中的数据库URL为测试库地址 alembic_cfg.set_main_option( "sqlalchemy.url", "postgresql://user:password@localhost/test_db" ) # 执行迁移到最新版本 command.upgrade(alembic_cfg, "head") yield # 测试完成后回滚到初始状态(可选) command.downgrade(alembic_cfg, "base")
方式二:用子进程执行迁移命令
如果更习惯命令行操作,也可以直接调用subprocess:
import subprocess import pytest @pytest.fixture(scope="session") def test_db_migrations(): subprocess.run( ["alembic", "upgrade", "head", "-x", "db_url=postgresql://user:password@localhost/test_db"], check=True ) yield subprocess.run( ["alembic", "downgrade", "base", "-x", "db_url=postgresql://user:password@localhost/test_db"], check=True )
2. 替换Strawberry字段中的数据库会话
Strawberry支持依赖注入,和FastAPI的逻辑一致,你可以通过依赖重载实现测试会话替换:
第一步:封装数据库会话依赖
先把主库和测试库的会话封装成可替换的依赖:
# db/deps.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session # 主库配置 MAIN_DB_URL = "postgresql://user:password@localhost/main_db" # 测试库配置 TEST_DB_URL = "postgresql://user:password@localhost/test_db" def get_main_db_session() -> Session: engine = create_engine(MAIN_DB_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() try: yield db finally: db.close() def get_test_db_session() -> Session: engine = create_engine(TEST_DB_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() try: yield db finally: db.close()
第二步:在Strawberry Resolver中注入依赖
在字段定义时通过strawberry.dependency引入会话:
import strawberry from sqlalchemy.orm import Session from db.deps import get_main_db_session from models import User @strawberry.type class Query: @strawberry.field def get_user(self, id: int, db: Session = strawberry.dependency(get_main_db_session)) -> User: return db.query(User).filter(User.id == id).first()
第三步:测试时重载依赖
在pytest fixture中替换Strawberry的依赖:
# conftest.py import pytest from strawberry.fastapi import GraphQLRouter from fastapi.testclient import TestClient from main import app, graphql_app from db.deps import get_main_db_session, get_test_db_session @pytest.fixture(scope="function") def client(): # 重载依赖,用测试库会话替换主库会话 graphql_app.override_dependency(get_main_db_session, get_test_db_session) client = TestClient(app) yield client # 测试完成后恢复原依赖 graphql_app.reset_overrides()
3. 使用Requests模块测试GraphQL端点
直接构造GraphQL请求的JSON数据,发送到端点即可:
def test_get_user(client): # 定义GraphQL查询 query = """ query GetUser($id: Int!) { getUser(id: $id) { id name email } } """ # 发送POST请求 response = client.post( "/graphql", json={ "query": query, "variables": {"id": 1} } ) # 断言结果 assert response.status_code == 200 data = response.json() assert "errors" not in data assert data["data"]["getUser"]["id"] == 1
如果不用FastAPI的TestClient,直接用原生requests库:
import requests def test_get_user(): query = """ query GetUser($id: Int!) { getUser(id: $id) { id name } } """ response = requests.post( "http://localhost:8000/graphql", json={ "query": query, "variables": {"id": 1} } ) assert response.status_code == 200 assert "errors" not in response.json()
内容的提问来源于stack exchange,提问作者iljuhenson
相关产品推荐
相关产品推荐

