DataHub Smoke Test 编写指南面向活跃 GMS 实例的 E2E 测试权威规范【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahubDataHub 的smoke-test/目录承载着一整套针对运行中 GMSMetadata Graph Service实例的端到端E2EAPI 测试。本文以 smoke-test/AGENTS.md 为骨架结合 smoke-test/README.md、smoke-test/conftest.py 等仓库源码系统讲解何时该写 smoke test、如何保证用例幂等与隔离、如何正确使用 marker 与公共工具函数、以及本地/CI 如何运行这套测试。读完本文你将具备为 DataHub 任何新功能编写高质量 smoke test 的完整能力也能在遇到并发 flaky 测试时快速定位根因。一、什么是 DataHub Smoke TestDataHub smoke test 是运行在真实运行的 GMS 之上的 pytest 端到端测试覆盖写入 → 搜索可见、公开 API 流程、线上回归防护、真实堆栈鉴权等只有活着的 GMS 才能验证的行为。它与普通的单元测试、摄入测试有着明确分工逻辑在进程内即可验证的validator、parser、in-process service→ 用普通单元测试涉及数据摄入的 → 放 metadata-ingestion/tests/后端单测/集成测试 → 放对应模块的src/test/。只有行为确实依赖 GMS 运行时例如写入了元数据后能否被搜索到、公开 API 流程无既有覆盖、需要保留的生产回归、活栈上的鉴权才值得新增一个 smoke test。从测试性质看这类用例运行在慢速、共享的堆栈上且常以 pytest-xdist 并行执行因此编写规范的核心目标只有一个让并行运行的用例互不干扰、稳定可复现。二、何时新增 / 何时禁止新增AGENTS.md 给出了非常明确的边界这也是评审 smoke test PR 的第一道检查应当新增的情况行为需要 GMS写入 → 搜索的闭环公开 API 流程且没有既有覆盖需要保留的生产回归用例活堆栈上的授权authz行为。禁止新增的情况逻辑属于 validator、parser 或 in-process service —— 这些放单元测试即可流程已有覆盖 —— 去扩展对应模块不要新增test_*_v2.py之类的重复文件同一调用的正/反例堆叠 —— 一条 happy path 就足够只有真正需要 GMS 的负例如 policy deny才放进这里无效输入、组合式 4xx、大小写/空白变体都不属于 smoke test 的职责。放置位置与命名新测试放在tests/feature/目录下而不是塞进巨大的test_e2e.py连接器相关测试放在 metadata-ingestion/tests/实体名使用通用占位名如my_db.my_schema.events严禁出现客户标识或 ticket ID。从仓库目录结构可以直观看到这套组织方式smoke-test/tests/下按功能划分为tags_and_terms/、containers/、policies/、lineage/、search/、authorization/等四十余个特性目录每个特性目录承载自己的data.json与测试模块。三、六条核心编写规则AGENTS.md 将编写规则浓缩为六条每条背后都有对应的源码实现支撑1. 幂等且顺序无关测试不能假设 GMS 是空的也不能依赖其他测试的清理结果。这意味着不要假设模块级共享的可变全局状态不要假设某个 URN 一定不存在其他并行模块可能正在创建/删除它。2. 每次运行使用唯一命名任何被创建、修改或删除的实体其 URN 都必须带上运行唯一后缀。硬编码共享 URN 在 xdist 并行下必然 flake。仓库为此提供了unique_suffix()见 smoke-test/tests/utils.py它返回uuid.uuid4().hex[:8]专门用于规避并行模块之间的 URN 碰撞。3. 失败也要清理清理必须通过 fixtureyield或测试内的try/finally完成且删除操作要包在try/except里绝不能让清理失败掩盖真正的断言失败。仓库在 smoke-test/conftest.py 提供的_ingest_cleanup_data_impl/_ingest_cleanup_unique_dataset_impl就是这一规则的封装。4. 不要篡改共享平台状态不得修改 admin 用户、默认 All Users 策略或其他共享平台状态除非该模块标记了global_policy_mutator此类模块会在 CI 串行执行见下文 marker 部分。5. 用 logger不用 print统一使用logger.info()logging.getLogger(__name__)保证日志可被 CI 收集、按模块过滤。6. 禁止 sleep使用 Read-After-Write 等待GMS 的写入是异步落盘的经 Kafka → MAE Consumer → Elasticsearch绝不允许time.sleep()。正确姿势是读后写验证用with_test_retry()装饰器见 smoke-test/tests/utils.py基于 tenacity 实现重试次数与间隔由DATAHUB_TEST_SLEEP_TIMES/DATAHUB_TEST_SLEEP_BETWEEN环境变量控制批量摄入/清理调用wait_for_writes_to_sync()在 smoke-test/tests/consistency_utils.py用于等待 Kafka lag 归零与搜索索引可见只关心单一存储时wait_for_writes_to_sync(mcp_onlyTrue)或(mae_onlyTrue)可跳过无关等待加速执行已知trace_id的链路走 Trace API注意TestSessionWrapper已经在每次 POST/PUT 后自动等待写入同步因此除非你是在断言搜索/索引结果否则不要额外再同步一次。四、认证与配置别碰os.getenv和localhost:8080会话与客户端认证会话统一使用auth_sessionsession 级 fixture见 smoke-test/conftest.pyGraphQL 客户端使用graph_clientsmoke-test/conftest.py底层是datahub.ingestion.graph.client.DataHubGraphserver 与 token 均取自auth_session需要额外用户时用make_step_actor_user()smoke-test/tests/utilities/multi_user.py不要自行写一次性注册逻辑。环境变量统一入口所有环境变量读取必须经过 smoke-test/tests/utilities/env_vars.py而不是散落各处的os.getenv或硬编码localhost:8080。该文件是 smoke-test 的环境变量中央注册表核心变量包括变量默认值用途DATAHUB_GMS_URLhttp://localhost:8080GMS 地址DATAHUB_GMS_TOKEN无GMS Bearer Token远程实例优先用 token 认证DATAHUB_FRONTEND_URLhttp://localhost:9002Frontend 地址ADMIN_USERNAME/ADMIN_PASSWORDdatahub/datahub本地登录认证DATAHUB_KAFKA_URLlocalhost:9092Kafka 地址DATAHUB_TEST_SLEEP_BETWEEN20重试间隔秒DATAHUB_TEST_SLEEP_TIMES3重试次数ELASTICSEARCH_REFRESH_INTERVAL_SECONDS3搜索索引可见性的兜底等待BATCH_COUNT/BATCH_NUMBER1/0CI 矩阵分片PYTEST_XDIST_WORKERS未设置xdist 并行 worker 数SMOKE_POLICY_PHASE未设置1非 mutator /2mutator 阶段其中DATAHUB_GMS_TOKEN的优先级很有意思token 优先于登录。build_auth_session()smoke-test/conftest.py先检查DATAHUB_GMS_TOKEN存在则直接构造 token 会话跳过登录往返适合远程实例否则走get_frontend_session()登录并动态签发短期 GMS token。认证与鉴权的几个坑TestSessionWrappersmoke-test/tests/utils.py拦截 requests 的get/post/put等调用自动注入Authorization: Bearer token并在 POST/PUT 后自动执行同步等待它还支持raw_post跳过自动等待用于连续写入的场景只有conftest.py会发布 bootstrap admin token 到DATAHUB_GMS_TOKEN环境变量——受限用户的 wrapper 绝不能覆盖它否则wait_for_writes_to_sync()的 lag 轮询会因权限不足而失败execute_graphql()smoke-test/tests/utils.py已经断言了响应体非空、data不为None、无errors键所以测试内只需断言自己关心的字段即可。摄入数据用ingest_file_via_rest()smoke-test/tests/utils.py把data.json摄入 GMS它内部创建file → datahub-rest的 Pipeline完成后自动等待写入同步并轮询 GraphQL 搜索直到摄入的 URN 可被搜到。五、Markers领域、优先级与全局策略变更者每个测试模块必须声明pytestmark格式如下源自 AGENTS.mdfrom tests.utilities.domains import Domain pytestmark pytest.mark.domain(Domain.CATALOG) # pytestmark pytest.mark.domain(Domain.CATALOG, Domain.INGESTION)Domain枚举定义在 smoke-test/tests/utilities/domains.py五个产品域为platform、observe、ingestion、ai、catalog。Marker使用时机domain(...)必选。声明测试归属的产品域供--domain选择与 CI 按团队路由失败p0必须在每个 PR 上运行的测试global_policy_mutator会禁用默认策略或修改共享平台策略的测试CI 在并行模块之后串行运行它们此外pytest.mark.dependency()链尽量短理想 ≤3避免过深的依赖图拖慢整体执行。global_policy_mutator的串行机制在 smoke-test/smoke.sh 中有完整实现每次批量测试被拆成两个 pytest 调用——阶段 1 以 xdist 并行跑非 mutator 测试阶段 2 串行跑 mutator 测试带--reruns 1 --reruns-delay 1。这种先并行后串行的设计保证了共享策略状态不会被并发修改。conftest 中的_apply_smoke_policy_phase_filtersmoke-test/conftest.py依据SMOKE_POLICY_PHASE环境变量在收集阶段过滤出对应阶段的用例。六、隔离Isolation并行模块共享一个 GMS 时的生存法则多个测试模块会在同一个 GMS 上并行运行pytest-xdist --distloadscope按模块分组到不同 worker因此名字唯一是第一优先级。优先使用官方 helper而不是自己写uuid或字符串替换Helper用途unique_suffix()任何实体键用户、域、PAT…的唯一后缀unique_dataset_urn(name)在代码中创建GraphQL / SDK的 dataset URNmaterialize_with_unique_name(src, name, dest_dir)重写 fixture 文件name必须只出现在 URN 键中而非描述等自由文本返回(dest_file, unique_name)materialize_unique_dataset(...)dataset fixture →(dest_file, dataset_urn)_ingest_cleanup_unique_dataset_impl(...)重写 摄入 yield URN 删除。不做预删除URN 是全新的是共享 dataset 名的默认方案_ingest_cleanup_data_impl(...)预删除 → 摄入 → 清理。仅当键已经对该模块唯一时使用标准用法模块级 fixtureAGENTS.md 给出的标准模板如下可直接复制from conftest import _ingest_cleanup_unique_dataset_impl pytest.fixture(scopemodule, autouseTrue) def dataset_urn(auth_session, graph_client, tmp_path_factory): yield from _ingest_cleanup_unique_dataset_impl( auth_session, graph_client, tests/tags_and_terms/data.json, tags_and_terms, test-tags-terms-sample-kafka, tmp_path_factory.mktemp(tags_and_terms), )测试函数以dataset_urn为 fixture 参数即可拿到唯一化的 dataset。多实体的 fixture 需要对每个键调用materialize_with_unique_name可参考 smoke-test/tests/containers/containers_test.py。测试中途创建的实体AGENTS.md 给出了中途创建的标准清理范式——try/finally中把删除包进try/exceptdataset_urn unique_dataset_urn(my-feature) try: graph_client.emit(...) wait_for_writes_to_sync() finally: try: delete_urn(graph_client, dataset_urn) except Exception: logger.warning(cleanup failed for %s, dataset_urn, exc_infoTrue)delete_urnsmoke-test/tests/utils.py对 structured property 会先软删除再硬删除保证清理幂等。七、公共工具函数总览AGENTS.md 最后梳理了三大工具模块编写新测试时应优先复用1. smoke-test/tests/utils.py核心工具execute_graphql— 执行 GraphQL 并内置错误断言ingest_file_via_rest— 文件摄入 等待索引delete_urn/delete_urns/delete_urns_from_file— 多形态删除with_test_retry— 读后写重试装饰器wait_for_writes_to_sync— 写入同步等待unique_suffix/unique_dataset_urn/materialize_with_unique_name/materialize_unique_dataset— 唯一命名与 fixture 重写wait_for_ingested_urns_searchable/wait_for_browse_path_entities— 轮询搜索/浏览直至可见TestSessionWrapper— 注入 Bearer token 并自动同步的请求会话。2. smoke-test/tests/utilities/concurrent_test_runner.pyrun_concurrent_tests/run_concurrent_tests_with_args— 线程安全的并发测试运行器用于在单个测试内模拟并发场景。3. smoke-test/tests/utilities/concurrent_openapi.pyrun_tests(auth_session, fixture_globs...)— 基于 JSON fixture 的并发 OpenAPI 测试fixture 格式为{request, response}配合 DeepDiff 的exclude_regex_paths忽略不稳定字段。注意不要添加只是重复验证已有 GraphQL 路径的 OpenAPI fixture。4. smoke-test/tests/utilities/metadata_operations.pyadd_tag/remove_tag/add_term/remove_term/update_description— 封装常见元数据变更的 GraphQL mutation避免各测试复制粘贴addTagget_search_results/search_across_lineage/scroll_across_lineage等 — 带重试的只读查询。八、如何运行 smoke test本地与 CI本地快速开始按 smoke-test/README.md 的指引1. 前置条件本地跑起 DataHub# 仓库根目录 ./gradlew quickstartDebug2. 一次性搭建 Python 环境# 仓库根目录安装 metadata-ingestion venv ./gradlew :metadata-ingestion:installDev # 配置 smoke-test 专用环境 cd smoke-test python3 -m venv venv source venv/bin/activate pip install --upgrade pip wheel setuptools pip install -r requirements.txt3. 设置环境变量并运行cd smoke-test source venv/bin/activate export DATAHUB_VERSIONv1.0.0rc3-SNAPSHOT # 或当前版本 export TEST_STRATEGYpytests # 全部测试慢需要完整环境 pytest -vv # 推荐按文件运行 pytest test_system_info.py -vv # 按方法运行 pytest test_system_info.py::test_system_info_main_endpoint -vv # 多个指定测试 pytest test_e2e.py::test_healthchecks test_e2e.py::test_gms_usage_fetch -v按领域选择--domain可重复# 单一领域 pytest --domain catalog -vv # 多领域属于任一领域的测试都会运行 pytest --domain catalog --domain ingestion -vv--domain的实现在 smoke-test/conftest.py 与 smoke-test/tests/utilities/domains.py未知领域值会直接抛出pytest.UsageError而不是静默匹配零测试——避免绿灯但什么都没测的假成功。按关键级别选择# 只跑 p0 级别 pytest -m p0 -vv # p0 级别 领域过滤 pytest -m p0 --domain catalog -vvCI 中的选择机制SMOKE_TIER环境变量驱动 tier 选择p0或fullCI 的docker-unified.yml在 PR 上通过PYTEST_P0_SMOKE仓库变量决定是否只跑 p0见 smoke-test/smoke.shPR 触碰的测试模块即使没有 p0 marker 也会被运行CI 设置SMOKE_CHANGED_TESTSconftest 在pytest_itemcollected阶段给这些模块注入 p0 markersmoke-test/conftest.py从而被-m p0选中当改动波及面超出自身模块如共享 fixture、conftest.py、大范围重构时给 PR 打上run-all-tests标签即可跑全量——注意该标签从事件 payload 读取重新运行既有 workflow 不会生效需要重新 push。测试分类速览System Info 测试test_system_info.py快约 30 秒可独立运行覆盖/openapi/v1/system-info、/properties、/spring-components三个端点Core E2E 测试test_e2e.py慢依赖完整摄入管线Kafka / Schema Registry首次摄入 fixture 失败是常见现象建议先跑单个方法。快速验证 API 是否就绪curl -s http://localhost:8080/health | head -5 curl -s http://localhost:8080/openapi/v1/system-info | jq . | head -20九、深入conftest 里的并行调度黑科技如果只是写测试掌握前八节已足够但理解 conftest 的调度逻辑能帮你调试为什么我的测试没跑/跑歪了领域过滤--domain在收集后、批处理前执行因此分片只打包真正会被运行的测试smoke-test/conftest.py失败重跑FILTERED_TESTSCI 重试时把失败模块列表写入文件conftest 只收集这些模块配合SMOKE_POLICY_PHASE过滤按耗时加权分片conftest 读取pytest_test_weights.json中每个测试的历史耗时按模块粒度做bin-packing 装箱bin_pack_taskssmoke-test/conftest.py。权重计算是阶段感知的phase_aware_module_weightsmoke-test/conftest.py把串行 mutator 测试的耗时乘以 xdist worker 数——因为串行一分钟的实际墙钟成本约等于并行 N 分钟。实测中若不这样做mutator 重的批次会系统性超时仓库记录到tests/authorization/test_aspect_write_auth.py单独就有约 7.6 分钟纯串行耗时两阶段执行smoke.sh用SMOKE_POLICY_PHASE1/2两次调用 pytestsmoke-test/smoke.sh阶段 1 并行、阶段 2 串行且两个阶段都收集 0 个测试会被判定为失败拒绝什么都没测还报绿JUnit 报告pytest_runtest_setup把 domain marker 写入 JUnit 的user_properties供 CI / PostHog 按团队路由失败smoke-test/conftest.py。另外 conftest 还有两个值得注意的保护机制每个测试前清空get_default_graph的 LRU 缓存防止run_datahub_cmd用到陈旧凭据以及会话级 fixture 校验 admin corpUser 的system/isSupportUser特权位在测试后未被篡改verify_admin_corpuser_info_unchanged。十、编写 checklist提交前逐项自查参照 AGENTS.md 的规则新测试合入前应通过以下检查测试放在tests/feature/未膨胀test_e2e.py行为确实需要 GMS写→搜、公开 API、回归、活栈 authz而非纯逻辑所有实体 URN 使用unique_suffix()系列 helper无硬编码共享 URN清理走 fixtureyield/try-finally删除包try/except失败不掩盖断言未修改 admin 用户、默认 All Users 策略除非模块标记global_policy_mutator使用logger.info()而非print()无time.sleep()读后写用with_test_retry()批量用wait_for_writes_to_sync()认证统一走auth_session/graph_client/make_step_actor_user()环境变量全部经由 smoke-test/tests/utilities/env_vars.pyGraphQL 只断言本测试关心的字段execute_graphql已内建错误断言声明了pytestmark pytest.mark.domain(...)按需补充p0/global_policy_mutator复用 smoke-test/tests/utils.py 与 smoke-test/tests/utilities/metadata_operations.py 的公共 helper没有复制粘贴addTag或自造 uuid。遵循这套规范写出的 smoke test在共享 GMS、xdist 并行、CI 分片重试的高压环境下依然能够稳定、幂等、快速收敛是 DataHub 持续集成质量门禁的可靠基石。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
DataHub Smoke Test 编写指南：面向活跃 GMS 实例的 E2E 测试权威规范
DataHub Smoke Test 编写指南面向活跃 GMS 实例的 E2E 测试权威规范【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahubDataHub 的smoke-test/目录承载着一整套针对运行中 GMSMetadata Graph Service实例的端到端E2EAPI 测试。本文以 smoke-test/AGENTS.md 为骨架结合 smoke-test/README.md、smoke-test/conftest.py 等仓库源码系统讲解何时该写 smoke test、如何保证用例幂等与隔离、如何正确使用 marker 与公共工具函数、以及本地/CI 如何运行这套测试。读完本文你将具备为 DataHub 任何新功能编写高质量 smoke test 的完整能力也能在遇到并发 flaky 测试时快速定位根因。一、什么是 DataHub Smoke TestDataHub smoke test 是运行在真实运行的 GMS 之上的 pytest 端到端测试覆盖写入 → 搜索可见、公开 API 流程、线上回归防护、真实堆栈鉴权等只有活着的 GMS 才能验证的行为。它与普通的单元测试、摄入测试有着明确分工逻辑在进程内即可验证的validator、parser、in-process service→ 用普通单元测试涉及数据摄入的 → 放 metadata-ingestion/tests/后端单测/集成测试 → 放对应模块的src/test/。只有行为确实依赖 GMS 运行时例如写入了元数据后能否被搜索到、公开 API 流程无既有覆盖、需要保留的生产回归、活栈上的鉴权才值得新增一个 smoke test。从测试性质看这类用例运行在慢速、共享的堆栈上且常以 pytest-xdist 并行执行因此编写规范的核心目标只有一个让并行运行的用例互不干扰、稳定可复现。二、何时新增 / 何时禁止新增AGENTS.md 给出了非常明确的边界这也是评审 smoke test PR 的第一道检查应当新增的情况行为需要 GMS写入 → 搜索的闭环公开 API 流程且没有既有覆盖需要保留的生产回归用例活堆栈上的授权authz行为。禁止新增的情况逻辑属于 validator、parser 或 in-process service —— 这些放单元测试即可流程已有覆盖 —— 去扩展对应模块不要新增test_*_v2.py之类的重复文件同一调用的正/反例堆叠 —— 一条 happy path 就足够只有真正需要 GMS 的负例如 policy deny才放进这里无效输入、组合式 4xx、大小写/空白变体都不属于 smoke test 的职责。放置位置与命名新测试放在tests/feature/目录下而不是塞进巨大的test_e2e.py连接器相关测试放在 metadata-ingestion/tests/实体名使用通用占位名如my_db.my_schema.events严禁出现客户标识或 ticket ID。从仓库目录结构可以直观看到这套组织方式smoke-test/tests/下按功能划分为tags_and_terms/、containers/、policies/、lineage/、search/、authorization/等四十余个特性目录每个特性目录承载自己的data.json与测试模块。三、六条核心编写规则AGENTS.md 将编写规则浓缩为六条每条背后都有对应的源码实现支撑1. 幂等且顺序无关测试不能假设 GMS 是空的也不能依赖其他测试的清理结果。这意味着不要假设模块级共享的可变全局状态不要假设某个 URN 一定不存在其他并行模块可能正在创建/删除它。2. 每次运行使用唯一命名任何被创建、修改或删除的实体其 URN 都必须带上运行唯一后缀。硬编码共享 URN 在 xdist 并行下必然 flake。仓库为此提供了unique_suffix()见 smoke-test/tests/utils.py它返回uuid.uuid4().hex[:8]专门用于规避并行模块之间的 URN 碰撞。3. 失败也要清理清理必须通过 fixtureyield或测试内的try/finally完成且删除操作要包在try/except里绝不能让清理失败掩盖真正的断言失败。仓库在 smoke-test/conftest.py 提供的_ingest_cleanup_data_impl/_ingest_cleanup_unique_dataset_impl就是这一规则的封装。4. 不要篡改共享平台状态不得修改 admin 用户、默认 All Users 策略或其他共享平台状态除非该模块标记了global_policy_mutator此类模块会在 CI 串行执行见下文 marker 部分。5. 用 logger不用 print统一使用logger.info()logging.getLogger(__name__)保证日志可被 CI 收集、按模块过滤。6. 禁止 sleep使用 Read-After-Write 等待GMS 的写入是异步落盘的经 Kafka → MAE Consumer → Elasticsearch绝不允许time.sleep()。正确姿势是读后写验证用with_test_retry()装饰器见 smoke-test/tests/utils.py基于 tenacity 实现重试次数与间隔由DATAHUB_TEST_SLEEP_TIMES/DATAHUB_TEST_SLEEP_BETWEEN环境变量控制批量摄入/清理调用wait_for_writes_to_sync()在 smoke-test/tests/consistency_utils.py用于等待 Kafka lag 归零与搜索索引可见只关心单一存储时wait_for_writes_to_sync(mcp_onlyTrue)或(mae_onlyTrue)可跳过无关等待加速执行已知trace_id的链路走 Trace API注意TestSessionWrapper已经在每次 POST/PUT 后自动等待写入同步因此除非你是在断言搜索/索引结果否则不要额外再同步一次。四、认证与配置别碰os.getenv和localhost:8080会话与客户端认证会话统一使用auth_sessionsession 级 fixture见 smoke-test/conftest.pyGraphQL 客户端使用graph_clientsmoke-test/conftest.py底层是datahub.ingestion.graph.client.DataHubGraphserver 与 token 均取自auth_session需要额外用户时用make_step_actor_user()smoke-test/tests/utilities/multi_user.py不要自行写一次性注册逻辑。环境变量统一入口所有环境变量读取必须经过 smoke-test/tests/utilities/env_vars.py而不是散落各处的os.getenv或硬编码localhost:8080。该文件是 smoke-test 的环境变量中央注册表核心变量包括变量默认值用途DATAHUB_GMS_URLhttp://localhost:8080GMS 地址DATAHUB_GMS_TOKEN无GMS Bearer Token远程实例优先用 token 认证DATAHUB_FRONTEND_URLhttp://localhost:9002Frontend 地址ADMIN_USERNAME/ADMIN_PASSWORDdatahub/datahub本地登录认证DATAHUB_KAFKA_URLlocalhost:9092Kafka 地址DATAHUB_TEST_SLEEP_BETWEEN20重试间隔秒DATAHUB_TEST_SLEEP_TIMES3重试次数ELASTICSEARCH_REFRESH_INTERVAL_SECONDS3搜索索引可见性的兜底等待BATCH_COUNT/BATCH_NUMBER1/0CI 矩阵分片PYTEST_XDIST_WORKERS未设置xdist 并行 worker 数SMOKE_POLICY_PHASE未设置1非 mutator /2mutator 阶段其中DATAHUB_GMS_TOKEN的优先级很有意思token 优先于登录。build_auth_session()smoke-test/conftest.py先检查DATAHUB_GMS_TOKEN存在则直接构造 token 会话跳过登录往返适合远程实例否则走get_frontend_session()登录并动态签发短期 GMS token。认证与鉴权的几个坑TestSessionWrappersmoke-test/tests/utils.py拦截 requests 的get/post/put等调用自动注入Authorization: Bearer token并在 POST/PUT 后自动执行同步等待它还支持raw_post跳过自动等待用于连续写入的场景只有conftest.py会发布 bootstrap admin token 到DATAHUB_GMS_TOKEN环境变量——受限用户的 wrapper 绝不能覆盖它否则wait_for_writes_to_sync()的 lag 轮询会因权限不足而失败execute_graphql()smoke-test/tests/utils.py已经断言了响应体非空、data不为None、无errors键所以测试内只需断言自己关心的字段即可。摄入数据用ingest_file_via_rest()smoke-test/tests/utils.py把data.json摄入 GMS它内部创建file → datahub-rest的 Pipeline完成后自动等待写入同步并轮询 GraphQL 搜索直到摄入的 URN 可被搜到。五、Markers领域、优先级与全局策略变更者每个测试模块必须声明pytestmark格式如下源自 AGENTS.mdfrom tests.utilities.domains import Domain pytestmark pytest.mark.domain(Domain.CATALOG) # pytestmark pytest.mark.domain(Domain.CATALOG, Domain.INGESTION)Domain枚举定义在 smoke-test/tests/utilities/domains.py五个产品域为platform、observe、ingestion、ai、catalog。Marker使用时机domain(...)必选。声明测试归属的产品域供--domain选择与 CI 按团队路由失败p0必须在每个 PR 上运行的测试global_policy_mutator会禁用默认策略或修改共享平台策略的测试CI 在并行模块之后串行运行它们此外pytest.mark.dependency()链尽量短理想 ≤3避免过深的依赖图拖慢整体执行。global_policy_mutator的串行机制在 smoke-test/smoke.sh 中有完整实现每次批量测试被拆成两个 pytest 调用——阶段 1 以 xdist 并行跑非 mutator 测试阶段 2 串行跑 mutator 测试带--reruns 1 --reruns-delay 1。这种先并行后串行的设计保证了共享策略状态不会被并发修改。conftest 中的_apply_smoke_policy_phase_filtersmoke-test/conftest.py依据SMOKE_POLICY_PHASE环境变量在收集阶段过滤出对应阶段的用例。六、隔离Isolation并行模块共享一个 GMS 时的生存法则多个测试模块会在同一个 GMS 上并行运行pytest-xdist --distloadscope按模块分组到不同 worker因此名字唯一是第一优先级。优先使用官方 helper而不是自己写uuid或字符串替换Helper用途unique_suffix()任何实体键用户、域、PAT…的唯一后缀unique_dataset_urn(name)在代码中创建GraphQL / SDK的 dataset URNmaterialize_with_unique_name(src, name, dest_dir)重写 fixture 文件name必须只出现在 URN 键中而非描述等自由文本返回(dest_file, unique_name)materialize_unique_dataset(...)dataset fixture →(dest_file, dataset_urn)_ingest_cleanup_unique_dataset_impl(...)重写 摄入 yield URN 删除。不做预删除URN 是全新的是共享 dataset 名的默认方案_ingest_cleanup_data_impl(...)预删除 → 摄入 → 清理。仅当键已经对该模块唯一时使用标准用法模块级 fixtureAGENTS.md 给出的标准模板如下可直接复制from conftest import _ingest_cleanup_unique_dataset_impl pytest.fixture(scopemodule, autouseTrue) def dataset_urn(auth_session, graph_client, tmp_path_factory): yield from _ingest_cleanup_unique_dataset_impl( auth_session, graph_client, tests/tags_and_terms/data.json, tags_and_terms, test-tags-terms-sample-kafka, tmp_path_factory.mktemp(tags_and_terms), )测试函数以dataset_urn为 fixture 参数即可拿到唯一化的 dataset。多实体的 fixture 需要对每个键调用materialize_with_unique_name可参考 smoke-test/tests/containers/containers_test.py。测试中途创建的实体AGENTS.md 给出了中途创建的标准清理范式——try/finally中把删除包进try/exceptdataset_urn unique_dataset_urn(my-feature) try: graph_client.emit(...) wait_for_writes_to_sync() finally: try: delete_urn(graph_client, dataset_urn) except Exception: logger.warning(cleanup failed for %s, dataset_urn, exc_infoTrue)delete_urnsmoke-test/tests/utils.py对 structured property 会先软删除再硬删除保证清理幂等。七、公共工具函数总览AGENTS.md 最后梳理了三大工具模块编写新测试时应优先复用1. smoke-test/tests/utils.py核心工具execute_graphql— 执行 GraphQL 并内置错误断言ingest_file_via_rest— 文件摄入 等待索引delete_urn/delete_urns/delete_urns_from_file— 多形态删除with_test_retry— 读后写重试装饰器wait_for_writes_to_sync— 写入同步等待unique_suffix/unique_dataset_urn/materialize_with_unique_name/materialize_unique_dataset— 唯一命名与 fixture 重写wait_for_ingested_urns_searchable/wait_for_browse_path_entities— 轮询搜索/浏览直至可见TestSessionWrapper— 注入 Bearer token 并自动同步的请求会话。2. smoke-test/tests/utilities/concurrent_test_runner.pyrun_concurrent_tests/run_concurrent_tests_with_args— 线程安全的并发测试运行器用于在单个测试内模拟并发场景。3. smoke-test/tests/utilities/concurrent_openapi.pyrun_tests(auth_session, fixture_globs...)— 基于 JSON fixture 的并发 OpenAPI 测试fixture 格式为{request, response}配合 DeepDiff 的exclude_regex_paths忽略不稳定字段。注意不要添加只是重复验证已有 GraphQL 路径的 OpenAPI fixture。4. smoke-test/tests/utilities/metadata_operations.pyadd_tag/remove_tag/add_term/remove_term/update_description— 封装常见元数据变更的 GraphQL mutation避免各测试复制粘贴addTagget_search_results/search_across_lineage/scroll_across_lineage等 — 带重试的只读查询。八、如何运行 smoke test本地与 CI本地快速开始按 smoke-test/README.md 的指引1. 前置条件本地跑起 DataHub# 仓库根目录 ./gradlew quickstartDebug2. 一次性搭建 Python 环境# 仓库根目录安装 metadata-ingestion venv ./gradlew :metadata-ingestion:installDev # 配置 smoke-test 专用环境 cd smoke-test python3 -m venv venv source venv/bin/activate pip install --upgrade pip wheel setuptools pip install -r requirements.txt3. 设置环境变量并运行cd smoke-test source venv/bin/activate export DATAHUB_VERSIONv1.0.0rc3-SNAPSHOT # 或当前版本 export TEST_STRATEGYpytests # 全部测试慢需要完整环境 pytest -vv # 推荐按文件运行 pytest test_system_info.py -vv # 按方法运行 pytest test_system_info.py::test_system_info_main_endpoint -vv # 多个指定测试 pytest test_e2e.py::test_healthchecks test_e2e.py::test_gms_usage_fetch -v按领域选择--domain可重复# 单一领域 pytest --domain catalog -vv # 多领域属于任一领域的测试都会运行 pytest --domain catalog --domain ingestion -vv--domain的实现在 smoke-test/conftest.py 与 smoke-test/tests/utilities/domains.py未知领域值会直接抛出pytest.UsageError而不是静默匹配零测试——避免绿灯但什么都没测的假成功。按关键级别选择# 只跑 p0 级别 pytest -m p0 -vv # p0 级别 领域过滤 pytest -m p0 --domain catalog -vvCI 中的选择机制SMOKE_TIER环境变量驱动 tier 选择p0或fullCI 的docker-unified.yml在 PR 上通过PYTEST_P0_SMOKE仓库变量决定是否只跑 p0见 smoke-test/smoke.shPR 触碰的测试模块即使没有 p0 marker 也会被运行CI 设置SMOKE_CHANGED_TESTSconftest 在pytest_itemcollected阶段给这些模块注入 p0 markersmoke-test/conftest.py从而被-m p0选中当改动波及面超出自身模块如共享 fixture、conftest.py、大范围重构时给 PR 打上run-all-tests标签即可跑全量——注意该标签从事件 payload 读取重新运行既有 workflow 不会生效需要重新 push。测试分类速览System Info 测试test_system_info.py快约 30 秒可独立运行覆盖/openapi/v1/system-info、/properties、/spring-components三个端点Core E2E 测试test_e2e.py慢依赖完整摄入管线Kafka / Schema Registry首次摄入 fixture 失败是常见现象建议先跑单个方法。快速验证 API 是否就绪curl -s http://localhost:8080/health | head -5 curl -s http://localhost:8080/openapi/v1/system-info | jq . | head -20九、深入conftest 里的并行调度黑科技如果只是写测试掌握前八节已足够但理解 conftest 的调度逻辑能帮你调试为什么我的测试没跑/跑歪了领域过滤--domain在收集后、批处理前执行因此分片只打包真正会被运行的测试smoke-test/conftest.py失败重跑FILTERED_TESTSCI 重试时把失败模块列表写入文件conftest 只收集这些模块配合SMOKE_POLICY_PHASE过滤按耗时加权分片conftest 读取pytest_test_weights.json中每个测试的历史耗时按模块粒度做bin-packing 装箱bin_pack_taskssmoke-test/conftest.py。权重计算是阶段感知的phase_aware_module_weightsmoke-test/conftest.py把串行 mutator 测试的耗时乘以 xdist worker 数——因为串行一分钟的实际墙钟成本约等于并行 N 分钟。实测中若不这样做mutator 重的批次会系统性超时仓库记录到tests/authorization/test_aspect_write_auth.py单独就有约 7.6 分钟纯串行耗时两阶段执行smoke.sh用SMOKE_POLICY_PHASE1/2两次调用 pytestsmoke-test/smoke.sh阶段 1 并行、阶段 2 串行且两个阶段都收集 0 个测试会被判定为失败拒绝什么都没测还报绿JUnit 报告pytest_runtest_setup把 domain marker 写入 JUnit 的user_properties供 CI / PostHog 按团队路由失败smoke-test/conftest.py。另外 conftest 还有两个值得注意的保护机制每个测试前清空get_default_graph的 LRU 缓存防止run_datahub_cmd用到陈旧凭据以及会话级 fixture 校验 admin corpUser 的system/isSupportUser特权位在测试后未被篡改verify_admin_corpuser_info_unchanged。十、编写 checklist提交前逐项自查参照 AGENTS.md 的规则新测试合入前应通过以下检查测试放在tests/feature/未膨胀test_e2e.py行为确实需要 GMS写→搜、公开 API、回归、活栈 authz而非纯逻辑所有实体 URN 使用unique_suffix()系列 helper无硬编码共享 URN清理走 fixtureyield/try-finally删除包try/except失败不掩盖断言未修改 admin 用户、默认 All Users 策略除非模块标记global_policy_mutator使用logger.info()而非print()无time.sleep()读后写用with_test_retry()批量用wait_for_writes_to_sync()认证统一走auth_session/graph_client/make_step_actor_user()环境变量全部经由 smoke-test/tests/utilities/env_vars.pyGraphQL 只断言本测试关心的字段execute_graphql已内建错误断言声明了pytestmark pytest.mark.domain(...)按需补充p0/global_policy_mutator复用 smoke-test/tests/utils.py 与 smoke-test/tests/utilities/metadata_operations.py 的公共 helper没有复制粘贴addTag或自造 uuid。遵循这套规范写出的 smoke test在共享 GMS、xdist 并行、CI 分片重试的高压环境下依然能够稳定、幂等、快速收敛是 DataHub 持续集成质量门禁的可靠基石。【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
