Skip to content

数据库迁移与测试 ​

数据库迁移(Alembic) ​

迁移原理与环境 ​

项目使用 Alembic 管理关系型数据库的 Schema 演进,核心文件位于 backend/alembic/:

  • alembic.ini:Alembic 主配置,script_location 指向 alembic/,迁移脚本模板等。
  • alembic/env.py:迁移运行环境。它在加载时通过 ImportUtil.find_models(MappedBase) 自动扫描工程中全部 model.py / models.py 的 ORM 模型,并将 MappedBase.metadata 作为 target_metadata,用于 autogenerate 比对。数据库URL 取自 settings.ASYNC_DB_URI,并使用 异步引擎(asyncmy / asyncpg)执行迁移。
  • alembic/versions/:存放按时间顺序串联的迁移脚本(通过 down_revision / revision 形成链表)。
  • alembic/script.py.mako:迁移文件模板,预置 upgrade() / downgrade() 骨架。

所有 ORM 模型都继承自 app.framework.database.model.MappedBase,新增模型只要放在模块的 model.py 中就会被迁移自动感知,无需手动登记。

迁移命令 ​

项目在 backend/main.py 中封装了 Typer 命令行工具,推荐直接使用:

bash
# 进入后端目录
cd backend

# 生成迁移脚本(比对模型与库结构,自动产出增量变更)
uv run python main.py revision --env=dev

# 应用所有未执行的迁移到最新版本(head)
uv run python main.py upgrade --env=dev

--env 对应运行环境(如 dev / prod),决定加载哪个 .env.<环境> 配置。

也可以直接使用原生 Alembic CLI(需配置好环境变量或 .env):

命令作用
uv run alembic revision --autogenerate -m "msg"自动比对模型与库表,生成迁移脚本
uv run alembic upgrade head执行全部未应用迁移
uv run alembic downgrade -1回退一个版本
uv run alembic history查看迁移历史
uv run alembic current查看当前版本

迁移工作流 ​

推荐的标准操作流程:

  1. 修改模型:在对应业务模块的 model.py 中调整字段、表注释或索引。
  2. 生成脚本:运行 python main.py revision,Alembic 会输出一个新的 versions/*.py 文件,并在其中标注 # ### commands auto generated by Alembic - please adjust! ###。
  3. (可选)人工复核:自动生成的脚本(尤其是涉及数据类型变更、枚举、命名约束)可能需要手工修正,务必检查 upgrade() / downgrade() 是否对称且可逆。
  4. 应用迁移:本地执行 python main.py upgrade 验证;上线前通过部署脚本统一执行。

迁移文件结构 ​

每个迁移脚本遵循统一结构,关键是两个函数:

  • upgrade():定义向前变更(新增表、列、索引、约束等)。
  • downgrade():定义回滚操作,必须与 upgrade() 严格对称。

文件头部的元数据决定迁移顺序:

  • revision:本版本唯一标识。
  • down_revision:上一版本标识,形成单链表;首个迁移为 None。

下面是一个典型迁移片段(对已有表加列并建立索引):

python
def upgrade() -> None:
    op.add_column(
        "bpm_process_instance",
        sa.Column("business_key", sa.BigInteger(), nullable=True, comment="业务单据编号"),
    )
    op.create_index(
        op.f("ix_bpm_process_instance_business_key"),
        "bpm_process_instance",
        ["business_key"],
        unique=False,
    )

def downgrade() -> None:
    op.drop_index(op.f("ix_bpm_process_instance_business_key"), table_name="bpm_process_instance")
    op.drop_column("bpm_process_instance", "business_key")

单元测试(pytest) ​

测试依赖与配置 ​

测试依赖(pytest、pytest-asyncio)定义在 backend/pyproject.toml 的 dev 依赖组中。相关配置如下:

toml
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["."]
asyncio_mode = "auto"          # 异步用例无需手动标记
asyncio_default_fixture_loop_scope = "session"

使用 uv 安装开发依赖后,即可在 backend 目录直接运行 pytest。

测试数据库隔离 ​

测试框架把数据库与开发/生产数据完全隔离:

  • tests/conftest.py 在导入任何 app 模块之前覆盖环境变量,将 DATABASE_NAME 改为 mars-ai-studio-test(环境变量优先级高于 .env 文件)。
  • 各模块(如 tests/bpm/conftest.py、tests/framework/conftest.py)的 _ensure_database() 会在首次运行时:
    1. 通过管理员连接执行 CREATE DATABASE IF NOT EXISTS ... CHARACTER SET utf8mb4 创建独立测试库;
    2. 使用 MappedBase.metadata.create_all 依据 ORM 元数据直接建表(而非执行 Alembic 迁移)。

注意:测试环境的表结构来自 create_all,与迁移脚本是两条独立路径。迁移是生产环境 Schema 的权威来源;测试则通过元数据快速重建结构,便于本地验证。两者应保持一致——修改模型后,应同时生成迁移并在测试中验证。

测试夹具(fixtures) ​

常用夹具由 conftest.py 提供,避免重复搭建上下文:

  • db:每个用例独立的异步 AsyncSession。用例结束后会 TRUNCATE 全部表兜底,保证用例间互不污染。
  • make_auth:构造指定用户编号的 AuthSchema 认证上下文(默认 user_id=1),用于模拟登录用户。
  • make_engine(BPM):基于 make_auth 构造流程引擎 BpmProcessEngine,直接调用引擎逻辑,不走 HTTP 层。
  • make_definition / get_running_task(BPM):构造流程定义、查询运行中的任务等专用夹具。

编写测试用例 ​

测试聚焦框架与业务流程本身(如 CRUDBase 查询、BPM 引擎推进),不依赖 HTTP 层。由于 asyncio_mode = "auto",异步用例直接写成 async def 即可:

python
import pytest
from app.api.v1.module_infra.file.crud import InfraFileCRUD


async def test_known_column_filter_applies(db, make_auth):
    """真实列键条件正常生效:仅命中匹配记录"""
    crud = InfraFileCRUD(make_auth())
    await crud.create(data={"name": "a.txt", "path": "dir/a.txt", "size": 1, "type": "text/plain"})

    matched = await crud.get_list(search={"name": "a.txt"})
    assert len(matched) == 1
    assert matched[0].name == "a.txt"

要点:

  • 用 make_auth() 构造操作主体,make_engine() 构造可执行业务的引擎。
  • 通过 db 会话写入数据后,断言查询结果;用例结束自动清空,无需手动清理。
  • 通过 pytest.raises(...) 验证异常分支(如非法查询字段应抛出 CustomException)。

运行测试 ​

bash
# 运行全部用例
pytest

# 运行指定模块
pytest tests/framework
pytest tests/bpm

# 运行单个文件或用例
pytest tests/framework/test_crud_search.py
pytest tests/framework/test_crud_search.py::test_known_column_filter_applies

前端测试(Vitest) ​

前端基于 Vue Vben Admin,使用 Vitest 做单元测试(挂载环境为 happy-dom),命令在 frontend/package.json 中:

bash
# 运行一次单元测试(含 DOM 环境)
pnpm test:unit

# 监听模式
pnpm vitest

测试配置文件为 frontend/vitest.config.ts。业务代码(如 apps/web-ele)与通用包(packages/)可分别组织 .test.ts / .spec.ts 用例,覆盖工具函数、Store 逻辑与组件渲染。典型用例使用 @vue/test-utils 挂载组件并断言行为:

ts
import { mount } from '@vue/test-utils'
import { describe, expect, it } from 'vitest'

import MyButton from './my-button.vue'

describe('MyButton', () => {
  it('渲染插槽内容', () => {
    const wrapper = mount(MyButton, { slots: { default: '提交' } })
    expect(wrapper.text()).toContain('提交')
  })
})

端到端测试 ​

目录 tests/agent-browser 提供基于 agent-browser 的浏览器端到端自动化测试示例,用于覆盖关键用户路径。运行方式参见该目录说明。