Prefect 测试套件深入解析数据库 Fixture 设计原理与编写最佳实践【免费下载链接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.项目地址: https://gitcode.com/GitHub_Trending/pr/prefect在 Prefect 这个 Python 工作流编排框架中测试不仅要覆盖引擎、任务与流程运行还要验证服务端的状态机、编排规则与数据模型。为了保证这些测试既快又可靠Prefect 构建了一套以SQLite 与 PostgreSQL 双数据库为底座、以数据库 Fixture 为核心资产的测试体系。本文以 tests/README.md 为主干结合 tests/conftest.py、tests/fixtures/database.py 等源码讲解数据库 Fixture 的设计动机、编写规范、底层生命周期以及如何正确使用clear_db标记、事件断言等进阶技巧读完即可上手为 Prefect或同类使用 SQLAlchemy pytest-asyncio 的项目编写高质量的数据库级测试。一、测试体系概览SQLite 与 Postgres 双轨运行tests/README.md开篇即点明核心约束测试同时针对 SQLite 和 PostgreSQL 运行。这意味着任何依赖数据库行为的测试——从数据模型、编排规则到 API 路由——都必须在这两种引擎下表现一致。从 tests/AGENTS.md 可以看到更完整的结构约定tests/目录结构与src/prefect/一一对应tests/server/、tests/client/、tests/cli/、tests/engine/等数据库相关测试在 CI 中会同时跑两套引擎测试必须确定性不依赖时序、排序或外部状态测试数据要优先使用真实对象能用真实 flow、deployment、flow run 就不用 mockunittest.mock仅保留给外部服务和时间敏感操作。这一“真实优先”的原则正是数据库 Fixture 存在的意义——它们负责在测试运行前把真实、合法的领域对象写进数据库。二、数据库 Fixture 的核心作用原文要点数据库 Fixture位于./fixtures/database.py在测试运行前操纵数据库状态。它们最常见的用途是向数据库填充有用信息这样测试就不必手动创建每个需要的对象。在 Prefect 中服务端模型的测试通常需要一条 flow、一次 flow run、一个 deployment 才能开始。如果每个测试都从models.flows.create_flow开始手动搭建代码会大量重复且容易在参数细节上出错。数据库 Fixture 把“造数据”这件事集中封装测试只需声明依赖就能拿到已提交到数据库的真实 ORM 对象。从实现看tests/fixtures/database.py 提供了成体系的“数据工厂”覆盖了 Prefect 服务端的核心领域对象Fixture创建的数据库对象关键依赖flowFlowflow 记录sessionflow_runFlowRun Flowsession,flowtask_runTaskRun FlowRunsession,flow_rundeploymentDeployment含调度、存储文档、参数 Schemasession,flow,flow_function,storage_document_id,work_queue_1work_queue/work_poolWorkQueue / WorkPoolsessionblock_type_x/y/zBlockTypesessionblock_schema/block_documentBlockSchema / BlockDocumentsession,block_type_x等concurrency_limit_v2ConcurrencyLimitV2全局并发限制sessionlogs三条不同 level、不同 name 的 Log 记录session,log_datainitialize_orchestration返回 Flow/TaskOrchestrationContext用于编排规则测试session,flow正如 tests/fixtures/AGENTS.md 总结的服务端测试的典型 Fixture 链是session → flow → flow_run → task_run每个 Fixture 都在数据库中创建一个 ORM 对象并提交commit像deployment这种更复杂的对象则继续依赖flow、work_queue_1、storage_document_id等上游 Fixture 串成一条依赖链。三、编写数据库 Fixture 的四条最佳实践原文档给出的四条最佳实践每一条背后都有明确的工程动机。结合源码逐条展开1. 使用全局sessionfixture所有数据库 Fixture 都应把session作为参数。session在 tests/conftest.py 中并不定义它来自 tests/fixtures/database.py 的sessionfixturepytest.fixture async def session(db) - AsyncGenerator[AsyncSession, None]: session await db.session() async with session: yield session它通过provide_database_interface()拿到PrefectDBInterface并创建一个异步 SQLAlchemyAsyncSession。测试以及调用models.*的 Fixture共享同一会话作用域保证所有写入在同一个事务上下文内可被后续对象读取。2. 使用内部models函数创建或操纵对象Fixture 内部应使用 Prefect 服务端的内部模型函数prefect.server.models而不是直接操作 ORM 或绕过领域逻辑。例如创建 flow 调用models.flows.create_flow、创建 flow run 调用models.flow_runs.create_flow_run、创建任务运行调用models.task_runs.create_task_run。这样能保证测试数据经过与生产代码相同的校验与默认值处理路径数据形态与真实运行一致。3. 提交会话commit提交会话会让改动对“其他参与者”可见。Prefect 服务端测试中一个测试方法内部往往还需要通过models进一步查询或操纵数据只有先await session.commit()写入的对象才有持久化 ID 与状态才能被后续读取。4. 返回相关数据库对象Fixture 的返回值就是测试要用的对象。约定俗成地返回 ORM model如model调用方直接使用其.id、.name、.state等属性即可。完整示例原文档示例 源码还原原文档给出了flowFixture 的示例pytest.fixture async def flow(session): model await models.flows.create_flow( sessionsession, flowschemas.actions.FlowCreate(namemy-flow), ) await session.commit() return model而仓库中实际生效的版本tests/fixtures/database.py做了两处关键增强值得学习pytest.fixture async def flow(session: AsyncSession): model await models.flows.create_flow( sessionsession, flowschemas.core.Flow(namefmy-flow-{uuid.uuid4()}) ) await session.commit() return modelUUID 随机化命名fmy-flow-{uuid.uuid4()}保证每个测试拿到的 flow 名称全局唯一。这是为了避免跨测试、跨 workerpytest-xdist的命名冲突——这也是后面clear_db标记存在的前提完整类型注解session: AsyncSession标注参数类型符合 tests/AGENTS.md 中“所有测试函数与 Fixture 必须完整类型注解”的规范Python 3.12 风格如dict[str, str]。再来看一个更复杂的 Fixture 例子——flow_runpytest.fixture async def flow_run(session: AsyncSession, flow: orm_models.Flow): model await models.flow_runs.create_flow_run( sessionsession, flow_runschemas.core.FlowRun(flow_idflow.id, flow_version0.1), ) await session.commit() return model它声明依赖flow把上游 Fixture 返回的flow.id作为外键写入 FlowRun完整体现了 Fixture 依赖链的用法。四、数据库基础设施源码级解析数据库 Fixture 之所以开箱即用是因为 tests/fixtures/database.py 和 tests/conftest.py 里有一套完整的“数据库生命周期管理”设施。1. 数据库接口与会话作用域pytest.fixture(scopesession, autouseTrue) def db(test_database_connection_url, safety_check_settings): return provide_database_interface()db是 session 级自动 fixture提供统一的PrefectDBInterface。autouseTrue意味着整个测试会话自动生效无需显式声明依赖。2. 引擎创建与优雅销毁database_enginefixture 负责创建引擎并在会话结束时销毁所有引擎而不只是自己创建的那一个pytest.fixture(scopesession, autouseTrue) async def database_engine(db: PrefectDBInterface): Produce a database engine TRACKER.active True engine await db.engine() yield engine engines list(ENGINES.values()) ENGINES.clear() for engine in engines: await engine.dispose() TRACKER.clear() gc.collect()源码注释解释了原因测试期间可能有其他事件循环创建了额外引擎必须在会话结束后统一dispose()并主动清理连接引用否则会触发ResourceWarning未关闭连接等告警。3. 建库与清库pytest.fixture(scopesession, autouseTrue) async def setup_db(database_engine, db): Create all database objects prior to running tests, and drop them when tests are done. try: await db.create_db() # 测试前建库 yield except Exception as exc: raise RuntimeError(fFailed to set up the database at {database_engine.url!r}) from excsetup_db在测试前调用db.create_db()创建全部表结构失败时抛出带数据库 URL 的运行时错误方便定位环境问题。4. 会话级隔离临时 PREFECT_HOME 与测试 Profiletests/conftest.py 的pytest_sessionstart钩子会在测试会话启动时创建临时目录作为PREFECT_HOME避免污染真实环境建立名为test-session的 Profile关闭所有后台服务scheduler、late runs、event persister、triggers、task run recorder、analytics、heartbeats 等以消除测试期间的后台干扰并加速套件设置PREFECT_LOGGING_LEVELDEBUG、禁用 CLI 彩色输出便于断言等。配套的safety_check_settings会断言PREFECT_API_URL为 None——测试绝不允许连接到外部 API这是防止误操作的重要安全网。5. 双引擎下的连接 URL 生成tests/conftest.py 的generate_test_database_connection_url展示了双数据库测试的底层机制SQLite数据库文件随临时PREFECT_HOME隔离每个 xdist worker 生成独立的prefect_{worker_id}.dbPostgreSQL在连接 URL 基础上为每个 worker 创建独立测试库{database}_tests_{worker_id}测试结束前自动DROP DATABASE若因连接占用无法删除ObjectInUseError也会在下一次运行时先删除再重建保证可重入。五、clear_db标记按需清库的性能开关这是 tests/AGENTS.md 重点强调、而 README 未展开的关键约定与数据库 Fixture 强相关测试默认拿不到干净的数据库。如果一个测试依赖数据库“从空开始”——比如统计行数、列出“所有 X”、断言某记录不存在——必须显式加pytest.mark.clear_db标记。pytest.mark.clear_db # 单个测试 async def test_count_flows(): ... class TestFlowAPI: # 整个类 pytestmark pytest.mark.clear_db pytestmark pytest.mark.clear_db # 整个模块为什么默认不清库带 UUID 随机名的对象如my-flow-{uuid.uuid4()}、按 ID 过滤、只与自己创建的对象交互的测试天然不冲突跳过清库可带来25%–100% 的套件提速因此默认不清理如何检查标记是否多余运行uv run pytest tests/path/to/file.py --no-clear-db -x若测试仍通过即可删除pytestmark或单测标记提交。--no-clear-db选项定义在 tests/conftest.py 的pytest_addoption中正是为此审计场景而设清理机制的健壮性clear_dbfixturetests/fixtures/database.py在清理时按reversed(orm_models.Base.metadata.sorted_tables)逆序删除所有表先子表后父表规避外键约束并对InterfaceError/DBAPIError做最多 3 次重试SQLite 并发访问可能触发 “database is locked”同时还会清空内存版并发租约存储ConcurrencyLeaseStorage防止测试间污染。六、客户端与服务端 Fixture 的契约边界tests/fixtures/AGENTS.md 明确了三类 Fixture 的分工这是编写测试前必须选对的第一件事场景使用 Fixture底层形态客户端侧测试SDK 行为prefect_client/sync_prefect_client完整 SDK 客户端服务端 API 测试路由client/test_client/sync_client原始 httpx / FastAPI TestClient直连临时服务器见 tests/fixtures/api.py 的create_app(ephemeralTrue)服务端模型 / 编排测试session 预置 ORM Fixture异步 SQLAlchemy 会话关键契约服务端与客户端 Fixture 不得混用。服务端测试应使用精简客户端避免在服务端测试中使用完整 SDK 的prefect_client。这条边界保证了 API 测试聚焦 HTTP 契约、模型测试聚焦数据库与状态机互不干扰。七、编排规则测试initialize_orchestration实战对于服务端最核心的“编排规则”orchestration rules测试initialize_orchestration是无可替代的关键 Fixturetests/fixtures/database.py。它接受run_typeflow或task、初始状态类型、提议状态类型等参数内部完成创建 flow 与 flow run必要时附带 task run通过commit_flow_run_state/commit_task_run_state写入初始状态forceTrue构造FlowOrchestrationContext或TaskOrchestrationContext绑定初始状态与提议状态返回上下文供规则执行。典型用法见 tests/server/orchestration/test_rules.pyinitial_state_type states.StateType.PENDING proposed_state_type states.StateType.RUNNING intended_transition (initial_state_type, proposed_state_type) ctx await initialize_orchestration(session, run_type, *intended_transition) async with contextlib.AsyncExitStack() as stack: the_polite_hero PoliteHeroRule(ctx, *intended_transition) ctx await stack.enter_async_context(the_polite_hero) await ctx.validate_proposed_state() assert ctx.proposed_state_type states.StateType.COMPLETED assert ctx.validated_state_type states.StateType.COMPLETED这套机制让测试能够以极小的样板代码覆盖“PENDING → RUNNING → COMPLETED”等任意状态转换下的规则行为包括规则对状态的变更、超时拒绝变更、等待提议等边界场景。八、事件断言与 Flaky 测试治理在服务端测试中创建领域对象flow、work pool、variable、block、并发限制等会自动发出生命周期事件prefect.object.created等。因此凡是对AssertingEventsClient的总数或顺序做断言的测试必须在 Fixture 搭建完成后调用AssertingEventsClient.reset()否则会把“搭建 Fixture 时产生的事件”误计入被测动作导致断言失真。针对时序敏感的断言tests/AGENTS.md 推荐使用prefect._internal.testing的retry_assertsasync for attempt in retry_asserts(max_attempts5, delay0.5): with attempt: callback.assert_called_once_with(flow_run_id)另外两点容易踩坑的约定Flow 超时测试必须用线程超时pytest-timeout默认使用 SIGALRM会与 Prefect 自身的 SIGALRM 超时机制冲突需显式标记pytest.mark.timeout(methodthread)PREFECT_HOME只创建一次pytest_sessionstart不会在测试间重置直接向其中写文件如 PID 文件的测试必须在 Fixture teardown 中删除否则后续测试会把它当成真实状态。九、常用命令速查综合 tests/AGENTS.md 与仓库实际用法常用命令如下项目使用uv管理依赖uv run pytest tests/ # 运行全部测试SQLite uv run pytest tests/path.py -k test_name # 运行指定测试 uv run pytest tests/module/ -n4 # 4 进程并行pytest-xdist uv run pytest tests/path.py -x # 失败即停 uv run pytest tests/path.py -x --tbshort # 紧凑回溯 uv run pytest tests/path/to/file.py --no-clear-db -x # 审计 clear_db 标记是否多余PostgreSQL 下运行需要设置PREFECT_SERVER_DATABASE_CONNECTION_URL指向可用的 Postgres 实例测试用户需具备建库权限tests/conftest.py 会为每个 worker 自动创建/删除独立测试库。除此之外pytest_addoption还提供了服务筛选选项--exclude-services、--only-service、--only-services与--disable-docker-image-builds用于 CI 中对服务集成测试和 Docker 镜像构建做精细控制。十、总结Prefect 的测试体系以“SQLite PostgreSQL 双引擎”为地基以数据库 Fixture 为核心生产力工具沉淀出了一套清晰的设计原则Fixture 即数据工厂用session 内部models函数 commit 返回对象的四步法把“造数据”收敛为可复用的声明式依赖默认不清库 按需clear_db以 UUID 随机命名保证测试隔离换取 25%–100% 的套件提速严格的三层 Fixture 契约SDK 客户端、HTTP 客户端、SQLAlchemy 会话各司其职禁止混用真实优先、确定性优先尽量用真实对象mock 只留给外部服务所有结论都可通过 tests/fixtures/database.py、tests/conftest.py、tests/AGENTS.md 等文件验证。理解并复刻这套数据库 Fixture 设计不仅有助于为 Prefect 贡献高质量测试对于任何基于 SQLAlchemy pytest-asyncio 的工程而言也是一套可直接借鉴的数据库测试范式。【免费下载链接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.项目地址: https://gitcode.com/GitHub_Trending/pr/prefect创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Prefect 测试套件深入解析：数据库 Fixture 设计原理与编写最佳实践
Prefect 测试套件深入解析数据库 Fixture 设计原理与编写最佳实践【免费下载链接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.项目地址: https://gitcode.com/GitHub_Trending/pr/prefect在 Prefect 这个 Python 工作流编排框架中测试不仅要覆盖引擎、任务与流程运行还要验证服务端的状态机、编排规则与数据模型。为了保证这些测试既快又可靠Prefect 构建了一套以SQLite 与 PostgreSQL 双数据库为底座、以数据库 Fixture 为核心资产的测试体系。本文以 tests/README.md 为主干结合 tests/conftest.py、tests/fixtures/database.py 等源码讲解数据库 Fixture 的设计动机、编写规范、底层生命周期以及如何正确使用clear_db标记、事件断言等进阶技巧读完即可上手为 Prefect或同类使用 SQLAlchemy pytest-asyncio 的项目编写高质量的数据库级测试。一、测试体系概览SQLite 与 Postgres 双轨运行tests/README.md开篇即点明核心约束测试同时针对 SQLite 和 PostgreSQL 运行。这意味着任何依赖数据库行为的测试——从数据模型、编排规则到 API 路由——都必须在这两种引擎下表现一致。从 tests/AGENTS.md 可以看到更完整的结构约定tests/目录结构与src/prefect/一一对应tests/server/、tests/client/、tests/cli/、tests/engine/等数据库相关测试在 CI 中会同时跑两套引擎测试必须确定性不依赖时序、排序或外部状态测试数据要优先使用真实对象能用真实 flow、deployment、flow run 就不用 mockunittest.mock仅保留给外部服务和时间敏感操作。这一“真实优先”的原则正是数据库 Fixture 存在的意义——它们负责在测试运行前把真实、合法的领域对象写进数据库。二、数据库 Fixture 的核心作用原文要点数据库 Fixture位于./fixtures/database.py在测试运行前操纵数据库状态。它们最常见的用途是向数据库填充有用信息这样测试就不必手动创建每个需要的对象。在 Prefect 中服务端模型的测试通常需要一条 flow、一次 flow run、一个 deployment 才能开始。如果每个测试都从models.flows.create_flow开始手动搭建代码会大量重复且容易在参数细节上出错。数据库 Fixture 把“造数据”这件事集中封装测试只需声明依赖就能拿到已提交到数据库的真实 ORM 对象。从实现看tests/fixtures/database.py 提供了成体系的“数据工厂”覆盖了 Prefect 服务端的核心领域对象Fixture创建的数据库对象关键依赖flowFlowflow 记录sessionflow_runFlowRun Flowsession,flowtask_runTaskRun FlowRunsession,flow_rundeploymentDeployment含调度、存储文档、参数 Schemasession,flow,flow_function,storage_document_id,work_queue_1work_queue/work_poolWorkQueue / WorkPoolsessionblock_type_x/y/zBlockTypesessionblock_schema/block_documentBlockSchema / BlockDocumentsession,block_type_x等concurrency_limit_v2ConcurrencyLimitV2全局并发限制sessionlogs三条不同 level、不同 name 的 Log 记录session,log_datainitialize_orchestration返回 Flow/TaskOrchestrationContext用于编排规则测试session,flow正如 tests/fixtures/AGENTS.md 总结的服务端测试的典型 Fixture 链是session → flow → flow_run → task_run每个 Fixture 都在数据库中创建一个 ORM 对象并提交commit像deployment这种更复杂的对象则继续依赖flow、work_queue_1、storage_document_id等上游 Fixture 串成一条依赖链。三、编写数据库 Fixture 的四条最佳实践原文档给出的四条最佳实践每一条背后都有明确的工程动机。结合源码逐条展开1. 使用全局sessionfixture所有数据库 Fixture 都应把session作为参数。session在 tests/conftest.py 中并不定义它来自 tests/fixtures/database.py 的sessionfixturepytest.fixture async def session(db) - AsyncGenerator[AsyncSession, None]: session await db.session() async with session: yield session它通过provide_database_interface()拿到PrefectDBInterface并创建一个异步 SQLAlchemyAsyncSession。测试以及调用models.*的 Fixture共享同一会话作用域保证所有写入在同一个事务上下文内可被后续对象读取。2. 使用内部models函数创建或操纵对象Fixture 内部应使用 Prefect 服务端的内部模型函数prefect.server.models而不是直接操作 ORM 或绕过领域逻辑。例如创建 flow 调用models.flows.create_flow、创建 flow run 调用models.flow_runs.create_flow_run、创建任务运行调用models.task_runs.create_task_run。这样能保证测试数据经过与生产代码相同的校验与默认值处理路径数据形态与真实运行一致。3. 提交会话commit提交会话会让改动对“其他参与者”可见。Prefect 服务端测试中一个测试方法内部往往还需要通过models进一步查询或操纵数据只有先await session.commit()写入的对象才有持久化 ID 与状态才能被后续读取。4. 返回相关数据库对象Fixture 的返回值就是测试要用的对象。约定俗成地返回 ORM model如model调用方直接使用其.id、.name、.state等属性即可。完整示例原文档示例 源码还原原文档给出了flowFixture 的示例pytest.fixture async def flow(session): model await models.flows.create_flow( sessionsession, flowschemas.actions.FlowCreate(namemy-flow), ) await session.commit() return model而仓库中实际生效的版本tests/fixtures/database.py做了两处关键增强值得学习pytest.fixture async def flow(session: AsyncSession): model await models.flows.create_flow( sessionsession, flowschemas.core.Flow(namefmy-flow-{uuid.uuid4()}) ) await session.commit() return modelUUID 随机化命名fmy-flow-{uuid.uuid4()}保证每个测试拿到的 flow 名称全局唯一。这是为了避免跨测试、跨 workerpytest-xdist的命名冲突——这也是后面clear_db标记存在的前提完整类型注解session: AsyncSession标注参数类型符合 tests/AGENTS.md 中“所有测试函数与 Fixture 必须完整类型注解”的规范Python 3.12 风格如dict[str, str]。再来看一个更复杂的 Fixture 例子——flow_runpytest.fixture async def flow_run(session: AsyncSession, flow: orm_models.Flow): model await models.flow_runs.create_flow_run( sessionsession, flow_runschemas.core.FlowRun(flow_idflow.id, flow_version0.1), ) await session.commit() return model它声明依赖flow把上游 Fixture 返回的flow.id作为外键写入 FlowRun完整体现了 Fixture 依赖链的用法。四、数据库基础设施源码级解析数据库 Fixture 之所以开箱即用是因为 tests/fixtures/database.py 和 tests/conftest.py 里有一套完整的“数据库生命周期管理”设施。1. 数据库接口与会话作用域pytest.fixture(scopesession, autouseTrue) def db(test_database_connection_url, safety_check_settings): return provide_database_interface()db是 session 级自动 fixture提供统一的PrefectDBInterface。autouseTrue意味着整个测试会话自动生效无需显式声明依赖。2. 引擎创建与优雅销毁database_enginefixture 负责创建引擎并在会话结束时销毁所有引擎而不只是自己创建的那一个pytest.fixture(scopesession, autouseTrue) async def database_engine(db: PrefectDBInterface): Produce a database engine TRACKER.active True engine await db.engine() yield engine engines list(ENGINES.values()) ENGINES.clear() for engine in engines: await engine.dispose() TRACKER.clear() gc.collect()源码注释解释了原因测试期间可能有其他事件循环创建了额外引擎必须在会话结束后统一dispose()并主动清理连接引用否则会触发ResourceWarning未关闭连接等告警。3. 建库与清库pytest.fixture(scopesession, autouseTrue) async def setup_db(database_engine, db): Create all database objects prior to running tests, and drop them when tests are done. try: await db.create_db() # 测试前建库 yield except Exception as exc: raise RuntimeError(fFailed to set up the database at {database_engine.url!r}) from excsetup_db在测试前调用db.create_db()创建全部表结构失败时抛出带数据库 URL 的运行时错误方便定位环境问题。4. 会话级隔离临时 PREFECT_HOME 与测试 Profiletests/conftest.py 的pytest_sessionstart钩子会在测试会话启动时创建临时目录作为PREFECT_HOME避免污染真实环境建立名为test-session的 Profile关闭所有后台服务scheduler、late runs、event persister、triggers、task run recorder、analytics、heartbeats 等以消除测试期间的后台干扰并加速套件设置PREFECT_LOGGING_LEVELDEBUG、禁用 CLI 彩色输出便于断言等。配套的safety_check_settings会断言PREFECT_API_URL为 None——测试绝不允许连接到外部 API这是防止误操作的重要安全网。5. 双引擎下的连接 URL 生成tests/conftest.py 的generate_test_database_connection_url展示了双数据库测试的底层机制SQLite数据库文件随临时PREFECT_HOME隔离每个 xdist worker 生成独立的prefect_{worker_id}.dbPostgreSQL在连接 URL 基础上为每个 worker 创建独立测试库{database}_tests_{worker_id}测试结束前自动DROP DATABASE若因连接占用无法删除ObjectInUseError也会在下一次运行时先删除再重建保证可重入。五、clear_db标记按需清库的性能开关这是 tests/AGENTS.md 重点强调、而 README 未展开的关键约定与数据库 Fixture 强相关测试默认拿不到干净的数据库。如果一个测试依赖数据库“从空开始”——比如统计行数、列出“所有 X”、断言某记录不存在——必须显式加pytest.mark.clear_db标记。pytest.mark.clear_db # 单个测试 async def test_count_flows(): ... class TestFlowAPI: # 整个类 pytestmark pytest.mark.clear_db pytestmark pytest.mark.clear_db # 整个模块为什么默认不清库带 UUID 随机名的对象如my-flow-{uuid.uuid4()}、按 ID 过滤、只与自己创建的对象交互的测试天然不冲突跳过清库可带来25%–100% 的套件提速因此默认不清理如何检查标记是否多余运行uv run pytest tests/path/to/file.py --no-clear-db -x若测试仍通过即可删除pytestmark或单测标记提交。--no-clear-db选项定义在 tests/conftest.py 的pytest_addoption中正是为此审计场景而设清理机制的健壮性clear_dbfixturetests/fixtures/database.py在清理时按reversed(orm_models.Base.metadata.sorted_tables)逆序删除所有表先子表后父表规避外键约束并对InterfaceError/DBAPIError做最多 3 次重试SQLite 并发访问可能触发 “database is locked”同时还会清空内存版并发租约存储ConcurrencyLeaseStorage防止测试间污染。六、客户端与服务端 Fixture 的契约边界tests/fixtures/AGENTS.md 明确了三类 Fixture 的分工这是编写测试前必须选对的第一件事场景使用 Fixture底层形态客户端侧测试SDK 行为prefect_client/sync_prefect_client完整 SDK 客户端服务端 API 测试路由client/test_client/sync_client原始 httpx / FastAPI TestClient直连临时服务器见 tests/fixtures/api.py 的create_app(ephemeralTrue)服务端模型 / 编排测试session 预置 ORM Fixture异步 SQLAlchemy 会话关键契约服务端与客户端 Fixture 不得混用。服务端测试应使用精简客户端避免在服务端测试中使用完整 SDK 的prefect_client。这条边界保证了 API 测试聚焦 HTTP 契约、模型测试聚焦数据库与状态机互不干扰。七、编排规则测试initialize_orchestration实战对于服务端最核心的“编排规则”orchestration rules测试initialize_orchestration是无可替代的关键 Fixturetests/fixtures/database.py。它接受run_typeflow或task、初始状态类型、提议状态类型等参数内部完成创建 flow 与 flow run必要时附带 task run通过commit_flow_run_state/commit_task_run_state写入初始状态forceTrue构造FlowOrchestrationContext或TaskOrchestrationContext绑定初始状态与提议状态返回上下文供规则执行。典型用法见 tests/server/orchestration/test_rules.pyinitial_state_type states.StateType.PENDING proposed_state_type states.StateType.RUNNING intended_transition (initial_state_type, proposed_state_type) ctx await initialize_orchestration(session, run_type, *intended_transition) async with contextlib.AsyncExitStack() as stack: the_polite_hero PoliteHeroRule(ctx, *intended_transition) ctx await stack.enter_async_context(the_polite_hero) await ctx.validate_proposed_state() assert ctx.proposed_state_type states.StateType.COMPLETED assert ctx.validated_state_type states.StateType.COMPLETED这套机制让测试能够以极小的样板代码覆盖“PENDING → RUNNING → COMPLETED”等任意状态转换下的规则行为包括规则对状态的变更、超时拒绝变更、等待提议等边界场景。八、事件断言与 Flaky 测试治理在服务端测试中创建领域对象flow、work pool、variable、block、并发限制等会自动发出生命周期事件prefect.object.created等。因此凡是对AssertingEventsClient的总数或顺序做断言的测试必须在 Fixture 搭建完成后调用AssertingEventsClient.reset()否则会把“搭建 Fixture 时产生的事件”误计入被测动作导致断言失真。针对时序敏感的断言tests/AGENTS.md 推荐使用prefect._internal.testing的retry_assertsasync for attempt in retry_asserts(max_attempts5, delay0.5): with attempt: callback.assert_called_once_with(flow_run_id)另外两点容易踩坑的约定Flow 超时测试必须用线程超时pytest-timeout默认使用 SIGALRM会与 Prefect 自身的 SIGALRM 超时机制冲突需显式标记pytest.mark.timeout(methodthread)PREFECT_HOME只创建一次pytest_sessionstart不会在测试间重置直接向其中写文件如 PID 文件的测试必须在 Fixture teardown 中删除否则后续测试会把它当成真实状态。九、常用命令速查综合 tests/AGENTS.md 与仓库实际用法常用命令如下项目使用uv管理依赖uv run pytest tests/ # 运行全部测试SQLite uv run pytest tests/path.py -k test_name # 运行指定测试 uv run pytest tests/module/ -n4 # 4 进程并行pytest-xdist uv run pytest tests/path.py -x # 失败即停 uv run pytest tests/path.py -x --tbshort # 紧凑回溯 uv run pytest tests/path/to/file.py --no-clear-db -x # 审计 clear_db 标记是否多余PostgreSQL 下运行需要设置PREFECT_SERVER_DATABASE_CONNECTION_URL指向可用的 Postgres 实例测试用户需具备建库权限tests/conftest.py 会为每个 worker 自动创建/删除独立测试库。除此之外pytest_addoption还提供了服务筛选选项--exclude-services、--only-service、--only-services与--disable-docker-image-builds用于 CI 中对服务集成测试和 Docker 镜像构建做精细控制。十、总结Prefect 的测试体系以“SQLite PostgreSQL 双引擎”为地基以数据库 Fixture 为核心生产力工具沉淀出了一套清晰的设计原则Fixture 即数据工厂用session 内部models函数 commit 返回对象的四步法把“造数据”收敛为可复用的声明式依赖默认不清库 按需clear_db以 UUID 随机命名保证测试隔离换取 25%–100% 的套件提速严格的三层 Fixture 契约SDK 客户端、HTTP 客户端、SQLAlchemy 会话各司其职禁止混用真实优先、确定性优先尽量用真实对象mock 只留给外部服务所有结论都可通过 tests/fixtures/database.py、tests/conftest.py、tests/AGENTS.md 等文件验证。理解并复刻这套数据库 Fixture 设计不仅有助于为 Prefect 贡献高质量测试对于任何基于 SQLAlchemy pytest-asyncio 的工程而言也是一套可直接借鉴的数据库测试范式。【免费下载链接】prefectPrefect is a workflow orchestration framework for building resilient data pipelines in Python.项目地址: https://gitcode.com/GitHub_Trending/pr/prefect创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
