FastAPI 中的单元测试

2024-03-26

我有一个使用 FastAPI 开发的后端应用程序,使用 SQLModel(SQLAlchemy 和 Pydantic)并连接到 Postgres 数据库。 我有集成测试来测试我的端点是否可以与暂存 PG DB 正常工作。 但现在我必须编写单元测试,并且我不知道如何继续测试我的端点和以隔离方式调用的函数。


这是我的项目的一个非常简化的版本:

The 我的项目的架构:(考虑每个文件夹下都有一个 __ init__.py 文件)

app/
├── api/
│   ├── core/
│   │   ├── config.py   #get the env settings and distribute it to the app
│   │   ├── .env
│   ├── crud/
│   │   ├── items.py    #the CRUD functions called by the router
│   ├── db/
│   │   ├── session.py  #the get_session function handling the db engine
│   ├── models/
│   │   ├── items.py    #the SQLModel object def as is in the db
│   ├── routers/
│   │   ├── items.py    #the routing system
│   ├── schemas/
│   │   ├── items.py    #the python object def as it is used in the app
│   ├── main.py         #the main app
├── tests/              #the pytest tests
│   ├── unit_tests/
│   ├── integration_tests/
│   │   ├── test_items.py

In the 增删改查/items.py:

from fastapi.encoders import jsonable_encoder
from sqlmodel import Session, select
from api.models import Item
from api.schemas import ItemCreate


def get_item(db_session: Session, item_id: int) -> Item:
    query = select(Item).where(Item.id == item_id)
    return db_session.exec(query).first()


def create_new_item(db_session: Session, *, obj_input: ItemCreate) -> Item:
    obj_in_data = jsonable_encoder(obj_input)
    db_obj = Item(**obj_in_data)
    db_session.add(db_obj)
    db_session.commit()
    db_session.refresh(db_obj)
    return db_obj

In the 数据库/会话.py:

from sqlalchemy.engine import Engine
from sqlmodel import create_engine, Session
from api.core.config import settings

engine: Engine = create_engine(settings.SQLALCHEMY_DATABASE_URI, pool_pre_ping=True)


def get_session() -> Session:
    with Session(engine) as session:
        yield session

In the 模型/items.py:

from sqlmodel import SQLModel, Field, MetaData

meta = MetaData(schema="pouetpouet")  # https://github.com/tiangolo/sqlmodel/issues/20


class Item(SQLModel, table=True):
    __tablename__ = "cities"
    # __table_args__ = {"schema": "pouetpouet"}
    metadata = meta

    id: int = Field(primary_key=True, default=None)
    city_name: str

In the 路由器/items.py:

from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session
from api.crud import get_item, create_new_item
from api.db.session import get_session
from api.models import Item
from api.schemas import ItemRead, ItemCreate

router = APIRouter(prefix="/api/items", tags=["Items"])


@router.get("/{item_id}", response_model=ItemRead)
def read_item(
    *,
    db_session: Session = Depends(get_session),
    item_id: int,
) -> Item:
    item = get_item(db_session=db_session, item_id=item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item


@router.post("/", response_model=ItemRead)
def create_item(
    *,
    db_session: Session = Depends(get_session),
    item_input: ItemCreate,
) -> Item:
    item = create_new_item(db_session=db_session, obj_input=item_input)
    return item

In the 模式/items.py:

from typing import Optional
from sqlmodel import SQLModel


class ItemBase(SQLModel):
    city_name: Optional[str] = None


class ItemCreate(ItemBase):
    pass

class ItemRead(ItemBase):
    id: int
    class Config:
        orm_mode: True

In the 测试/integration_tests/test_items.py:

from fastapi.testclient import TestClient
from api.main import app

client = TestClient(app)

def test_create_item() -> None:
    data = {"city_name": "Las Vegas"}
    response = client.post("/api/items/", json=data)
    assert response.status_code == 200
    content = response.json()
    assert content["city_name"] == data["city_name"]
    assert "id" in content

这里的重点是我觉得被困住了db_session: Sessioncrud/items.py 和 routers/items.py 中的所有函数中使用的参数,因为我认为必须获得有效 postgres 连接的有效会话以进行测试。


ps:在后端开发方面经验不足,如果您发现一些奇怪的情况,请毫不犹豫地对我的代码提出建设性意见。它将受到热烈欢迎。


正如我在评论中提到的你之前的问题 https://stackoverflow.com/questions/76530308, 我认为unit测试不应该处理以下数据库any种,而是模拟所有与数据库交互的函数。

建议的方法

这是我如何为 CRUD 函数编写单元测试的示例create_new_item.

首先是用于演示目的的简化设置:

from sqlmodel import Field, SQLModel, Session


class ItemCreate(SQLModel):
    x: str
    y: float


class Item(ItemCreate, table=True):
    id: int | None = Field(primary_key=True)


def create_new_item(db_session: Session, *, obj_input: ItemCreate) -> Item:
    db_obj = Item.from_orm(obj_input)
    db_session.add(db_obj)
    db_session.commit()
    db_session.refresh(db_obj)
    return db_obj

测试用例:

from unittest import TestCase
from unittest.mock import call, create_autospec

from sqlmodel import Session

# ... import Item, ItemCreate


class MyTestCase(TestCase):
    def test_create_new_item(self) -> None:
        test_item = ItemCreate(x="foo", y=3.14)
        mock_session = create_autospec(Session, instance=True)

        expected_output = Item.from_orm(test_item)
        expected_session_calls = [
            call.add(expected_output),
            call.commit(),
            call.refresh(expected_output),
        ]

        output = create_new_item(mock_session, obj_input=test_item)
        self.assertEqual(expected_output, output)
        self.assertListEqual(expected_session_calls, mock_session.mock_calls)

请注意,测试从未连接到任何数据库。我们must假设 SQLAlchemy 函数(以及 SQLModel 的扩展函数)如宣传的那样工作。否则我们一开始就不应该使用它们。我们的单元测试需要确保我们正确调用这些函数。


更大程度的隔离

从技术上来说,我们现在测试这个功能的方式,我们仍然依赖我们编写的其他代码以某种方式发挥作用,即模型。在我看来,这是一个灰色地带,因为从技术上讲,我们没有为这些类编写任何方法,而是依赖于基类的现有方法。

But you could认为模型定义本身应该与create_new_item功能。换句话说,单元测试的成功或失败不应该取决于我们的模型定义是否Item and ItemCreate match以一种允许一个从另一个构建的方式。

因此,具有最严格隔离的测试可能如下所示:

from unittest import TestCase
from unittest.mock import MagicMock, call, create_autospec, patch

from sqlmodel import Session

# ... import Item


class MyTestCase(TestCase):
    @patch.object(Item, "from_orm")
    def test_create_new_item2(self, mock_from_orm: MagicMock) -> None:
        test_item = MagicMock()
        mock_session = create_autospec(Session, instance=True)

        expected_output = mock_from_orm.return_value = object()
        expected_session_calls = [
            call.add(expected_output),
            call.commit(),
            call.refresh(expected_output),
        ]

        output = create_new_item(mock_session, obj_input=test_item)
        self.assertEqual(expected_output, output)
        self.assertListEqual(expected_session_calls, mock_session.mock_calls)

减少孤立:迈向一体化的一步

如果您想要进行单独类型的测试does执行数据库查询以检查结果是否符合预期,您可以注入绑定到 SQLite 内存数据库引擎的不同会话对象。

from unittest import TestCase

from sqlmodel import SQLModel, Session, create_engine, select

# ... import Item, ItemCreate


class MyTestCase(TestCase):
    def test_create_new_item3(self) -> None:
        engine = create_engine("sqlite:///")
        SQLModel.metadata.create_all(engine)

        test_item = ItemCreate(x="foo", y=3.14)
        expected_output = Item(**test_item.dict(), id=1)

        with Session(engine) as session:
            output = create_new_item(session, obj_input=test_item)
            self.assertEqual(expected_output, output)
            result = session.exec(select(Item)).all()
            self.assertListEqual([expected_output], result)

当然,如果您的模型具有与 SQLite 不兼容的特定于数据库的设置(例如 Postgres 架构),则这将不起作用。

在这种情况下,您有多种选择,如上一个问题中所述。例如,您实际上可以创建一个单独的测试数据库,并围绕测试编写一些隔离/清理逻辑。然后提供一个连接到该测试数据库的引擎。

或者您可以尝试巧妙地围绕这些特定设置进行猴子修补。

本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

FastAPI 中的单元测试 的相关文章

随机推荐

  • 尝试在 Javascript (ES5) 中实现 OPP 继承的简单方法

    只是出于好奇 我在 Javascript 中玩弄原型继承和 OOP 继承 大多数结果涉及用函数模拟 类 和 扩展 概念 而其他结果则使用原型和构造函数 我写了这段代码 function Warrior weaponName var weap
  • Android:“BadTokenException:无法添加窗口;您的活动正在运行吗?”在 PreferenceActivity 中显示对话框

    我想寻求一些帮助 在我的应用程序中 我只有一个活动 一个PreferenceActivity 不需要其他 它只是一个简单的后台同步应用程序 所以PrefsActivity是主 启动器 用户设置首选项后 检查checkBoxPreferenc
  • Java 和 PostgreSQL 之间的“坏记录 MAC”SSL 错误

    我们遇到了 Java 应用程序和 PostgreSQL 8 3 服务器之间随机断开连接的问题 并出现 坏记录 MAC SSL 错误 我们两边都运行 Debian Lenny 在客户端 我们看到 main WRITE TLSv1 Applic
  • 如何获取整个月的天数

    我正在创建全年日历 如何在 Flutter 中获取该月的天数 例如 一月 gt 31 二月 gt 28 29 按年 三月 gt 31 依此类推 Using date1 difference date2 inDays 是不正确的 对于某些日期
  • 从类和 javadoc 生成代码存根

    有人熟悉生成代码存根的工具吗具有有意义的名称来自类和javadoc 真正的问题应该是 我的类没有调试信息和匹配的 javadoc 但我的 IntelliJ IDEA 8 0 1 拜托 没有 IDE 战争 没有考虑 javadoc 并向我显示
  • 可以在单个进程中在 JNI 中创建多个 JVM 吗? [复制]

    这个问题在这里已经有答案了 我有一个在单个进程中运行并允许使用模块的 C 框架 我想添加的一种类型的模块是加载 JAR 然后调用其中特定的预定义函数的模块 根据用户的需要 可以有任意数量的模块同时运行 这就提出了一个问题 每个模块是否可以使
  • GCC NRVO/RVO 警告

    有没有warning 这让我们知道是否NRVO RVO执行与否 在GCC 我找到 fno elide constructors关掉NRVO RVO but NRVO RVO有其发生的条件 有时不发生 有必要知道是否NRVO RVO当额外的复
  • 与多个属性共享枚举声明值

    我想要一个具有多个属性的类 可以用数值保存工作日 summary weekday integer collection weekday integer 我想我可以将整数映射到值使用枚举 http edgeapi rubyonrails or
  • ActiveModel 序列化器:运行时有条件吗?

    我使用rails 5 0 1 和active model serializers 0 10 2 我想以某种方式有条件地序列化has many协会 class Question lt ApplicationRecord has many re
  • 如何创建或生成 .mlpd 文件以在 xamarin profiler 中检查应用程序性能

    你好 Xamarin 团队 我正在 Visual Studio 2017 社区中处理 Xamarin 表单 我想检查 Xamarin Profiler 中的应用程序性能 为此 我需要创建一个 mlpd 来检查 Xamarin Profile
  • 单击链接时使用 jQuery/Javascript 更改下拉选项

    假设我有以下链接和下拉菜单 a href contact Send a mail to mother a a href contact Send a mail to father a a href contact Send a mail t
  • 在 Windows 上非侵入式解锁文件

    有没有办法使用 Python 脚本在 Windows 上解锁文件 该文件被另一个进程独占锁定 我需要一个不终止或中断锁定过程的解决方案 我已经看过了门户锁 http code activestate com recipes 65203 一种
  • 在 Eclipse 中重新启动应用程序

    我正在使用 Eclipse Juno 开发一个独立服务器 不是战争 我将它作为 Java 应用程序从 Eclipse 运行 在进行一些代码更改后 我想停止当前正在运行的服务器并再次启动它 我每天这样做几十次 我目前的做法如下 转到 调试 选
  • 如何创建具有自定义外设和内存映射的 QEMU ARM 机器?

    我正在为 Cortex M3 cpu 编写代码 并且正在使用以下命令执行单元测试qemu arm二进制 现在一切都很好 但我想知道我是否能够使用测试整个系统qemu system arm 我的意思是 我想为 qemu 编写自定义 机器 我将
  • 使用RTL布局时如何通过命中测试获取树视图项?

    描述 在从右到左阅读模式 RTL 下有树视图 如何在仅知道单击坐标的情况下获取被单击的节点 这是一个插入类 它使树视图使用 RTL 显示 并包含一个单击处理程序 您可以在其中看到问题 unit Unit1 interface uses Wi
  • 连接充电器时 Android 上的自动开机

    我正在开发一个 Android 应用程序 需要不断地保持该应用程序始终运行 在手机始终连接到电源插座的情况下 如果手机电池耗尽 是否有可能在无需用户干预的情况下自动启动应用程序 即自动打开手机电源 无需按任何电源按钮 并在连接充电器后启动
  • ValidateRequest 错误还是 SQL Server Bug?

    我正在读这个article http software security sans org blog 2011 07 22 bypassing validaterequest in asp net 它说 该字符用值 uff1c 表示 如果将
  • MSGBOX 在 WSH/VBS 中的位置

    这是我的下一个问题 我希望有人可以帮助我 是否可以在 wsh vbs 中放置 msgbox 替代文本 http www 4freeimagehost com uploads a9b04cde0527 jpg http www 4freeim
  • VBA 使用架构文件提取数据

    我下面有这个代码 Option Explicit Sub MadMule2 Dim IE As InternetExplorer Dim el Dim els Dim colDocLinks As New Collection Dim Ti
  • FastAPI 中的单元测试

    我有一个使用 FastAPI 开发的后端应用程序 使用 SQLModel SQLAlchemy 和 Pydantic 并连接到 Postgres 数据库 我有集成测试来测试我的端点是否可以与暂存 PG DB 正常工作 但现在我必须编写单元测