数据目录数据治理数据血缘后端前端数据工程数据集成【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址https://gitcode.com/GitHub_Trending/da/datahub点击查看免费下载本文是 DataHub 开源仓库中smoke-test/目录下的官方 smoke test 编写指南smoke-test/CLAUDE.md 通过AGENTS.md引用的规范正文即 smoke-test/AGENTS.md的深度展开。它面向对运行中的 DataHub GMS 实例执行 pytest 端到端E2E测试的场景回答什么时候该写 smoke test、怎么写才能并行安全、如何保证幂等与清理可靠、如何用 marker 支撑 CI 分级执行等一系列问题。读完本文你将掌握 DataHub smoke-test 套件的核心约定、tests/utils.py与conftest.py中的关键 helper 用法、领域domain与分级p0标记体系以及完整的本地运行方式。一、什么是 DataHub Smoke Test定位与适用范围DataHub 的 smoke test 是运行在真实、存活的 GMSMetadata Graph Service之上的 E2E API 测试套件位于 smoke-test 目录。与单元测试、纯 Python 摄入测试相比它验证的是写进去 → 搜得到这条跨服务链路包括 GraphQL API、OpenAPI、认证、搜索索引、浏览路径等真实后端行为。1.1 何时应该新增 smoke test根据 smoke-test/AGENTS.md只有当行为确实需要 GMS 参与时才考虑 smoke test需要真实 GMS 的写入到搜索write → search链路现有测试覆盖之外的公开 API 流程需要长期守护的生产回归production regression需要 live 栈上的授权/鉴权行为live-stack authz。新增测试优先放入smoke-test/tests/feature/目录而不是堆进test_e2e.py连接器connector相关测试则放在metadata-ingestion/tests/。1.2 何时不应该写 smoke test规范明确列出三类不要写的场景逻辑属于 validator、parser 或进程内服务—— 这类纯逻辑应写单元测试放在被测代码旁流程已有覆盖—— 应该扩展现有模块而不是新增test_*_v2.py之类的平行文件同一调用的正/反变体堆叠—— 一个 happy path 就够负数场景invalid input、组合式 4xx、大小写/空白测试只有在确实需要 GMS例如 policy deny时才放入 smoke test。原因很直接smoke test 是慢速的共享栈测试配合 xdist 并行多一个用例就多一分 flake 与 CI 耗时。优先选择代码旁的单元测试、metadata-ingestion/tests/的摄入测试或后端集成测试是这套规范反复强调的原则。1.3 资源与性能定位从 smoke-test/README.md 可以看到分类策略test_system_info.py系统信息 API 测试约 30 秒可独立跑完而test_e2e.py核心 E2E依赖完整摄入流水线全套运行可长达 30 分钟以上。这种快慢分离直接决定了开发时的选择日常迭代用快用例合入前再跑完整套件。二、运行前的环境准备与基本命令2.1 前置条件本地已启动 DataHub从仓库根目录执行./gradlew quickstartDebug一次性搭建 Python 环境# 从项目根目录安装 metadata-ingestion 的开发版 venv ./gradlew :metadata-ingestion:installDev # 进入 smoke-test 目录创建独立 venv cd smoke-test python3 -m venv venv source venv/bin/activate pip install --upgrade pip wheel setuptools pip install -r requirements.txt2.2 环境变量export DATAHUB_VERSIONv1.0.0rc3-SNAPSHOT # 或当前版本 export TEST_STRATEGYpytests2.3 运行测试的基本命令cd smoke-test source venv/bin/activate # 运行全部测试耗时很长需要完整环境 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更推荐从仓库根目录通过scripts/dev/datahub-dev.sh test test-path运行README 指出这优于./gradlew quickstartDebug。2.4 按领域与分级筛选所有 smoke test 必须声明产品领域domain通过tests/utilities/domains.py中定义的Domain枚举platform、observe、ingestion、ai、catalog标记# 只跑某个领域 pytest --domain catalog -vv # 跨领域命中任一领域即运行 pytest --domain catalog --domain ingestion -vv跨领域测试需声明全部所属领域如pytest.mark.domain(Domain.CATALOG, Domain.INGESTION)。分级方面pytest.mark.p0标记的测试必须每次 PR 都跑pytest -m p0 -vv # 只跑 p0 分级 pytest -m p0 --domain catalog -vvCI 通过SMOKE_TIER环境变量驱动选择docker-unified.yml在PYTEST_P0_SMOKE仓库变量为 true 时对 PR 设置p0合入后的运行仍执行完整套件。PR 涉及到的测试模块即使没有p0标记conftest 也会在收集阶段将其标记为 p0 纳入-m p0改动共享 fixture 或conftest.py本身时若需要更广覆盖可在 PR 上添加run-all-tests标签以运行全部测试。三、编写规范核心幂等、隔离与清理smoke-test 套件在xdist 并行 共享 GMS下运行smoke.sh使用--distloadscope这是理解全部规则的出发点。3.1 铁律一览幂等且顺序无关不能假设 GMS 是空的也不能依赖其他测试的清理禁止模块级可变全局状态no module-level mutable globals。运行唯一命名每个创建、修改、删除的实体都要用运行唯一的名字见下节 Isolation。共享硬编码 URN 在 xdist 下必然 flake。失败也要清理用 fixtureyield或测试内try/finally清理删除要包try/except避免清理异常掩盖断言失败。不要改动共享平台状态admin 用户、默认 All Users 策略等除非模块标记了global_policy_mutator。日志用logger.info()而非print()。禁止time.sleep()等待 GMS/搜索/Kafka 收敛读后写一致用with_test_retry()批量摄入/清理用wait_for_writes_to_sync()仅一方相关的场景可加mcp_only/mae_only参数。已知trace_id场景用 Trace API。认证统一走 helperauth_session/graph_client额外用户用make_step_actor_user()配置一律从tests/utilities/env_vars.py读取而不是直接os.getenv或硬编码localhost:8080。复用工具函数不要复制粘贴addTag或自造uuidunique_suffix()已经存在标签/术语/描述操作统一走tests/utilities/metadata_operations.py。3.2 为什么TestSessionWrapper已经帮你等同步tests/utils.py中的TestSessionWrapper会拦截requests会话的POST/PUT并在写后自动调用wait_for_writes_to_sync()tests/utils.py。因此不要在每次写入后额外手动同步除非你正在断言的是搜索/索引结果。它同时自动注入 Bearer token批量连续写时可用raw_post/no_sync_waitTrue跳过每次的同步等待在批次结束后统一等一次。3.3 GraphQL 断言的正确姿势execute_graphql()tests/utils.py已经断言响应体非空、data不为None、且不存在errors键。规范强调只断言你关心的字段不要重复断言框架已保证的内容query query getDataset($urn: String!) { dataset(urn: $urn) { name } } res_data execute_graphql(auth_session, query, {urn: urn:li:dataset:(...)}) dataset_name res_data[data][dataset][name]3.4 认证与多用户管理员会话get_frontend_session()登录后得到auth_sessionGMS 只读操作可使用graph_client。额外用户通过tests/utilities/multi_user.py的make_step_actor_user(admin_session, prefix)创建原生用户并返回(urn, session)不要手写一次性注册流程。实现会先幂等删除同名用户、用 invite token 走/signUp、恢复 admin 的 frontend cookie再以新用户身份登录multi_user.py。配套cleanup_step_actor_user()负责清理。3.5 Marker 体系from tests.utilities.domains import Domain pytestmark pytest.mark.domain(Domain.CATALOG) # 跨领域pytestmark pytest.mark.domain(Domain.CATALOG, Domain.INGESTION)Marker何时使用domain(...)必填。取值platform、observe、ingestion、ai、catalog。p0必须每次 PR 运行。global_policy_mutator会禁用默认策略/修改共享平台策略。CI 在并行模块之后串行运行这些模块。--domain参数由conftest.py的pytest_addoption注册并在pytest_configure阶段校验取值非法值会以pytest.UsageError报错而不是静默匹配零测试conftest.py。领域标记还会被拷贝进 JUnituser_properties供 CI / PostHog 归属失败用例。另外保持pytest.mark.dependency()依赖链尽量短理想 ≤3。四、隔离策略运行唯一名字与 fixture 模板4.1 核心 helper 对照表Helper用途unique_suffix()任意实体键user、domain、PAT……追加短随机后缀基于uuid4().hex[:8]tests/utils.py。unique_dataset_urn(name)在代码中GraphQL / SDK创建 dataset 时生成唯一 URN默认平台kafka、环境PROD。materialize_with_unique_name(src, name, dest_dir)fixture 重写将name替换为name-suffix后写新文件返回(dest_file, unique_name)。name必须只出现在 URN 键中绝不能出现在 description 等自由文本里否则会被误替换。materialize_unique_dataset(...)dataset fixture →(dest_file, dataset_urn)保证文件内字段级引用 URN 与返回的 dataset URN 一致。_ingest_cleanup_unique_dataset_impl(...)重写 摄入 yield URN 清理。不预删除URN 每次全新是共享 dataset 名字时的默认选择。_ingest_cleanup_data_impl(...)预删除 → 摄入 → 清理。仅当键已经对该模块唯一时使用。4.2 标准 fixture 模板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 参数接收。多实体 fixture 需要对每个键调用materialize_with_unique_name参考tests/containers/containers_test.py。4.3 测试中途创建实体dataset_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)这里try/except嵌套体现了清理不能掩盖断言的原则即使清理失败也只记 warning保证原始断言结果可见。底层delete_urn对structuredProperty会先软删除再硬删除以保持幂等tests/utils.py。4.4 批量摄入与清理tests/utils.py提供ingest_file_via_rest(auth_session, filename)通过 datahub-rest sink 摄入文件默认ASYNC_BATCH模式完成后依次等待wait_for_writes_to_sync()与wait_for_ingested_urns_searchable()轮询 GraphQLsearchAcrossEntities直至摄入 URN 可被搜索到tests/utils.py。delete_urns_from_file()会按CLEANUP_DATA环境变量默认 true决定是否真正清理并用 10 并发 job 并行删除后统一同步。五、关键模块速查与源码路径围绕 smoke-test/AGENTS.md 提到的所有 helper可在以下文件中找到实现tests/utils.pyexecute_graphql、ingest_file_via_rest、delete_urn/delete_urns/delete_urns_from_file、with_test_retry、wait_for_writes_to_sync后者实现在tests/consistency_utils.py、unique_suffix、unique_dataset_urn、materialize_with_unique_name、materialize_unique_dataset、TestSessionWrappertests/utils.py。tests/utilities/concurrent_test_runner.py线程安全的run_concurrent_tests/run_concurrent_tests_with_args。tests/utilities/concurrent_openapi.pyrun_tests(auth_session, fixture_globs...)配合 JSON fixture{request, response}支持 DeepDiffexclude_regex_paths。不要添加只是重复覆盖已有 GraphQL 路径的 OpenAPI fixture。tests/utilities/env_vars.py所有环境变量的唯一读取入口env_vars.py包括DATAHUB_GMS_URL、ADMIN_USERNAME/ADMIN_PASSWORD默认datahub/datahub、DATAHUB_TEST_SLEEP_BETWEEN默认 20 秒/DATAHUB_TEST_SLEEP_TIMES默认 3 次、DB_TYPE等。tests/utilities/domains.pyDomain枚举与--domain选择逻辑domains.py。tests/utilities/multi_user.pymake_step_actor_user/cleanup_step_actor_user。tests/utilities/metadata_operations.pyadd_tag/remove_tag/add_term等标签、术语、描述操作。with_test_retry()的默认行为由get_sleep_info()即上面两个环境变量决定也可显式传max_attemptswith_test_retry() # 使用环境变量默认3 次、间隔 20 秒 with_test_retry(max_attempts10) # 最多重试 10 次它基于 tenacity 实现tests/utils.py是替代裸time.sleep的读后写一致性手段。execute_graphql底层使用TestSessionWrapper的post写后自动同步或raw_postno_sync_waitTrue时跳过保证断言前写入已收敛。六、CI 运行机制与本地调试要点6.1 CI 与本地差异CI./gradlew :smoke-test:pytest完整 Docker 流水线smoke.sh依据PYTEST_XDIST_WORKERS必须匹配^[1-9][0-9]*$env_vars.py决定是否传-n给 pytest并将global_policy_mutator模块串行到第二阶段的 policy phase 运行。本地直接 pytest使用本地运行的 DataHub 实例更快。建议开发期用本地最终合入前用 CI 验证。6.2 常见问题排查Connection refusedDataHub 未启动 / 端口不对GMS 应为 8080/ 服务仍在启动中多等几分钟。curl 直连返回 401属预期行为——测试套件自行处理认证请用测试套件访问受保护端点。Kafka/Schema Registry 连接错误只影响含摄入的完整 E2Etest_system_info.py仍可运行。建议改为运行单个测试方法。Python 环境异常重建 venv 后pip install -r requirements.txt。6.3 环境自检curl -s http://localhost:8080/health # GMS 健康检查 source venv/bin/activate pip list | grep datahub echo DATAHUB_VERSION: $DATAHUB_VERSION TEST_STRATEGY: $TEST_STRATEGY6.4 目录组织test_e2e.py—— 主 E2E 套件依赖完整摄入链路test_system_info.py—— 系统信息 API 测试/openapi/v1/system-info系列端点约 30 秒conftest.py—— 全局 fixture 与--domain等选项注册conftest.pytests/utils.py及tests/utilities/—— 上述全部 helper 的集中地smoke-test/tests/feature/—— 按功能划分的测试模块目录tags_and_terms/、containers/、authorization/、policies/等 50 个功能目录。七、总结一套面向真实后端的测试纪律DataHub smoke-test 规范的核心可以浓缩为三句话先判断值不值得写只有 GMS 参与的行为才写写则必须并行安全运行唯一命名 幂等 失败也清理标记驱动 CI 分级domain 归属 p0 门禁 global_policy_mutator 串行。在实际动手前建议通读 smoke-test/AGENTS.md 原文并对照 smoke-test/tests/utils.py 与 smoke-test/conftest.py 阅读 helper 实现即可在共享 GMS 的并行环境下写出稳定、可复用的 E2E 测试。赞分享数据目录数据治理数据血缘后端前端数据工程数据集成【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址https://gitcode.com/GitHub_Trending/da/datahub点击查看免费下载相关推荐如何轻松永久保存你的微信聊天记录留痕工具完整指南如何轻松永久保存你的微信聊天记录留痕工具完整指南 你是否曾因为手机更换而丢失珍贵的聊天记录是否希望将重要的对话转化为永恒的数字记忆在这个数字时代我们的数据目录数据治理数据血缘后端前端数据工程数据集成DataHub Smoke Test 规范实战基于 pytest 的端到端 API 测试标准与隔离、一致性、并发最佳实践DataHub Smoke Test 规范实战基于 pytest 的端到端 API 测试标准与隔离、一致性、并发最佳实践 导读 本篇文章以仓库 .agent数据目录数据治理数据血缘后端前端数据工程数据集成DataHub 冒烟测试模式参考面向实时 GMS 的 pytest E2E 测试实战指南DataHub 冒烟测试模式参考面向实时 GMS 的 pytest E2E 测试实战指南 本文以 DataHub 开源仓库 smoke test/ 冒烟测试套数据目录数据治理数据血缘后端前端数据工程数据集成上一篇推荐开源项目PEG —— 在Go中实现的Packrat解析表达式语法下一篇推荐开源项目Nodeunit——简洁高效的异步测试框架创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考