企业官网建设流程全解析

DataHub Notion 连接器集成测试实战：基于真实 Notion API 的端到端验证方案

【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub

本指南以 DataHub 仓库中 Notion 连接器的集成测试套件（metadata-ingestion/tests/integration/notion/）为核心，系统讲解如何在自己的 Notion 工作区中创建真实页面、配置测试凭据、运行与调试六类端到端测试场景，并深入解析连接器针对 unstructured-ingest v0.7.2 兼容性问题所应用的一系列 monkeypatch 源码实现。读完本文，你将掌握 Notion 连接器集成测试的完整前置准备、运行方式、断言逻辑与故障排查方法，同时理解连接器底层对 Notion API 新字段的容错处理机制。

一、这套集成测试在测什么

DataHub 的 Notion 连接器（源码位于 metadata-ingestion/src/datahub/ingestion/source/notion/notion_source.py）在摄取 Notion 页面时依赖第三方库 unstructured-ingest 0.7.2 完成页面解析。该库对 Notion API 的建模存在滞后：Notion API 陆续返回is_locked、description、list_start_index、list_format、新的 Icon 类型等字段，而 unstructured-ingest 0.7.2 的对应模型类并不认识它们，直接触发unexpected keyword argument、KeyError或ValueError等异常。

由于这些问题只有在真实页面与真实 API 响应下才会暴露，mock 无法复现，仓库因此在 metadata-ingestion/tests/integration/notion/test_notion_integration.py 中建立了一套调用真实 Notion API、创建真实页面、跑通完整摄取管道的集成测试，重点验证四个方面：

1. SyncBlock 兼容性—— 摄取包含"原始同步块 + 引用同步块"的页面，确认不再抛KeyError: 'children'；
2. NumberedListItem 支持—— 摄取包含编号列表项的页面，确认新 API 字段list_start_index、list_format被正确处理；
3. 凭据验证—— 对缺失的 AWS Bedrock / 无效的 Cohere 凭据给出早期醒目告警，同时保证文档仍被摄取（非阻塞设计）；
4. 完整摄取—— 一次性摄取所有测试页面，验证全部 monkeypatch 协同工作。

二、前置条件：创建 Notion Integration 与测试父页面

集成测试会在你的 Notion 工作区中真实创建页面，因此必须先完成两项配置。

1. 创建 Notion Integration

1. 打开 Notion 的 "My integrations" 管理页面；
2. 点击"+ New integration"；
3. 命名为 "DataHub Test Integration"（或任意名称）；
4. 选择目标工作区；
5. 启用Read content能力（测试需要读取页面与块内容）；
6. 点击Submit完成创建；
7. 复制Internal Integration Token（以secret_开头），即后续的NOTION_API_KEY。

2. 创建测试父页面

1. 在 Notion 中新建一个页面，命名为 "DataHub Test Pages"（或任意名称）；
2. 将该页面分享给测试 Integration：点击页面右上角"..."菜单 →"Add connections"→ 选择你的测试 Integration；
3. 从 URL 中提取页面 ID：URL 格式为https://www.notion.so/Page-Title-{PAGE_ID}，其中PAGE_ID是 32 位十六进制字符串（带不带连字符均可，脚本会自动去除连字符）。

若页面未与 Integration 分享，测试将报 "Page not found" 或 "Unauthorized" 错误，这是最常见的失败原因。

3. 一键式环境配置脚本

仓库提供了交互式配置脚本 metadata-ingestion/tests/integration/notion/setup_test_env.sh，它会：

• 校验NOTION_API_KEY是否以secret_开头；
• 校验父页面 ID 是否为 32 位十六进制（自动去除连字符）；
• 可选配置 AWS Bedrock 与 Cohere 凭据；
• 最终输出export ...命令，或写入权限为 600 的.env.notion_test文件供source使用。

三、运行集成测试

设置环境变量

# Required: Notion API credentials export NOTION_API_KEY="secret_abc123..." export NOTION_TEST_PARENT_PAGE_ID="abc123def456abc123def456abc123de" # Optional: For embedding credential tests export AWS_ACCESS_KEY_ID="your-aws-key" # For Bedrock tests export COHERE_API_KEY="your-cohere-key" # For Cohere tests

除此之外，测试文件还支持两个可选环境变量（详见 test_notion_integration.py 的模块注释）：

运行全部 Notion 集成测试

在仓库根目录执行：

cd metadata-ingestion ../gradlew :metadata-ingestion:testQuick -PtestFile=tests/integration/notion/test_notion_integration.py

或者激活虚拟环境后直接用 pytest：

pytest tests/integration/notion/test_notion_integration.py -v

运行特定测试

# 只测试 synced blocks pytest tests/integration/notion/test_notion_integration.py::test_notion_synced_blocks_ingestion -v # 只测试编号列表 pytest tests/integration/notion/test_notion_integration.py::test_notion_numbered_lists_ingestion -v # 只测试凭据告警 pytest tests/integration/notion/test_notion_integration.py::test_notion_bedrock_credential_warning -v

无凭据时的自动跳过机制

如果NOTION_API_KEY或NOTION_TEST_PARENT_PAGE_ID未设置，所有测试会被自动跳过并给出明确提示。该逻辑实现在 conftest.py 的pytest_collection_modifyitems钩子中：它检查环境变量，若不满足则为所有位于 notion 目录下的用例统一追加pytest.mark.skip标记：

SKIPPED [7] tests/integration/notion/conftest.py:16: NOTION_API_KEY and NOTION_TEST_PARENT_PAGE_ID environment variables must be set for Notion integration tests

这套机制让 CI 在未注入密钥时也能安全地跳过测试而不会失败。

四、Test Fixtures：session 级页面生命周期管理

测试页面由 conftest.py 中的 session 级 fixtures 统一创建，整个测试会话只创建一次，结束后自动清理。

页面组织结构

fixtures 会在工作区中建立一个清晰的层级：

Workspace Parent Page (NOTION_TEST_PARENT_PAGE_ID) └── DataHub Test Root Page (pytest {timestamp}) ├── Test Page - Synced Blocks (Auto-generated by pytest) ├── Test Page - Simple Content (Auto-generated by pytest) └── Test Page - Complex Content (Auto-generated by pytest)

核心 Fixtures 一览

• notion_client—— 使用NOTION_API_KEY实例化的认证 Notion API 客户端；
• notion_workspace_parent_page_id—— 读取NOTION_TEST_PARENT_PAGE_ID，作为测试根页面的父级；
• notion_test_root_page—— 创建名为 "DataHub Test Root Page" 的根页面（带时间戳），作为所有测试页面的容器；
• test_page_synced_blocks—— 包含原始同步块（synced_from: null，带子块）与引用同步块（synced_from: {block_id: ...}，无子块）的测试页；原始块创建后通过blocks.children.list找到其 block ID，再用blocks.children.append追加引用块；
• test_page_numbered_lists—— 含三个编号列表项的页面（"First/Second/Third numbered item"）；
• test_page_complex_content—— 混合多种块类型的页面（heading、paragraph、bulleted list、code、带 emoji 图标的 callout）；
• test_page_ids_for_ingestion—— 汇总上述三个测试页面 ID 的列表，供完整摄取测试使用。

清理机制

每个页面 fixture 在 yield 之后都会调用pages.update(page_id=..., archived=True)将页面归档（而非删除）。清理行为受should_cleanup_pages()控制（conftest.py）：当NOTION_TEST_CLEANUP为true时自动归档，否则保留页面并在终端打印访问链接。归档整个 Test Root Page 会级联归档其所有子页面。

若清理失败（如网络问题），可按以下标题手动归档或删除：

• "Test Page - Synced Blocks (Auto-generated by pytest)"
• "Test Page - Simple Content (Auto-generated by pytest)"
• "Test Page - Complex Content (Auto-generated by pytest)"

五、六大测试场景详解

所有测试都通过 DataHub 的Pipeline.create构造真实摄取管道（source 类型为notion，sink 为file），把产物写入临时 JSON 文件后做断言。下面逐一拆解。

1. SyncBlock Monkeypatch ——test_notion_synced_blocks_ingestion

管道配置（test_notion_integration.py）：

pipeline = Pipeline.create( { "run_id": "test-notion-synced-blocks", "source": { "type": "notion", "config": { "api_key": api_key, "page_ids": [page_id_to_test], "recursive": False, "advanced": { "continue_on_failure": False, "raise_on_error": True, }, }, }, "sink": {"type": "file", "config": {"filename": str(tmp_path / "notion_synced_blocks.json")}}, } )

该测试的核心断言是摄取不抛KeyError: 'children'，并验证日志中出现两处关键信息：

• "Applied monkeypatch to SyncBlock"（monkeypatch 确实被应用）；
• "synced blocks compatibility (original + reference)"（同时处理原始块与引用块）。

测试还会检查输出 JSON 是否包含页面标题与同步块内容（"ORIGINAL synced block content"、"Bullet point in synced block"）。需要说明的是：由于 unstructured-ingest v0.7.2 的限制，同步块内容可能无法完整提取，此时只要摄取过程无异常、monkeypatch 已生效，测试仍视为通过。

2. NumberedListItem Monkeypatch ——test_notion_numbered_lists_ingestion

针对 Notion API 在编号列表项上新增的list_start_index（起始序号，如从 2 开始）与list_format（numbers/letters/roman）字段，unstructured-ingest 0.7.2 会因不识别而抛TypeError。测试使用continue_on_failure: True、raise_on_error: False容忍部分失败，核心断言是日志出现"Applied monkeypatch to NumberedListItem"，并尽量验证三个列表项文本被摄取。

3. 完整摄取 ——test_notion_full_ingestion

一次性摄取全部测试页面，验证所有 monkeypatch 协同工作。它逐一断言日志中存在（test_notion_integration.py）：

• "Applied monkeypatch to SyncBlock"
• "Applied monkeypatch to NumberedListItem"
• "Applied monkeypatch to unstructured-ingest Page class"
• "database property classes"（即 "Applied monkeypatch to 22 database property classes"）

4. AWS Bedrock 凭据告警 ——test_notion_bedrock_credential_warning

先通过monkeypatch.delenv清除AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY、AWS_PROFILE，再配置embedding.provider: "bedrock"、model: "amazon.titan-embed-text-v1"。断言日志出现"WARNING: AWS Bedrock embeddings configured but credentials not detected"。该告警的生成逻辑位于 notion_source.py：当配置了 Bedrock 且环境变量、AWS profile、~/.aws/credentials均不存在时，输出一段醒目提示，说明文档仍会被摄取但语义搜索不可用。

5. Cohere 凭据错误 ——test_notion_cohere_credential_warning

配置embedding.provider: "cohere"、model: "embed-english-v3.0"、api_key: "invalid-key-for-testing"（故意使用无效凭据），从三个维度验证凭据错误检测（test_notion_integration.py）：

• 日志出现"EMBEDDING CREDENTIAL ERROR"；
• 或日志出现"EMBEDDING GENERATION FAILED FOR ALL ... DOCUMENTS"；
• 或NotionSourceReport中num_embedding_failures > 0且num_documents_with_embeddings == 0。

底层逻辑在 notion_source.py：捕获 embedding 异常后，若错误文本命中authfailure、credentials、unauthorized、invalid_api_key、accessdenied等关键词，即判定为凭据错误并以logger.error输出；且不记录文档状态，保证下次运行修复凭据后会自动重试。

6. 连接测试 ——test_notion_test_connection

直接调用NotionSource.test_connection(config)，断言report.basic_connectivity.capable为真，且能力报告中的"Page/Database Access"项capable为真，验证连接器对真实凭据的连接检测与页面访问上报能力。

六、monkeypatch 的源码级原理

所有 monkeypatch 都在 notion_source.py 中以_monkeypatch_*静态方法的形式实现，并统一遵循"导入目标类 → 保存原始方法 → 替换为补丁实现 → 日志确认"的模式。除测试重点验证的两个外，还包括：

其中 SyncBlock 补丁（notion_source.py）值得细看：当synced_from非空时，将数据交给DuplicateSyncedBlock.from_dict解析为引用块；当其为 null 且无children时，首次遇到会通过report.warning记录 "Synced Blocks Limitation"，提示同步块内容会因 v0.7.2 限制而缺失，但页面本身照常摄取。

对应的统计字段定义在 notion_report.py 的NotionSourceReport中：num_documents_with_embeddings、num_embedding_failures、embedding_failures、num_synced_blocks_skipped、synced_blocks_skipped等，供测试与运维诊断使用。

七、故障排查

测试全部被跳过

原因：环境变量未设置。解决：导出NOTION_API_KEY与NOTION_TEST_PARENT_PAGE_ID，并确认pytest_collection_modifyitems的跳过条件不再命中。

"Page not found" 或 "Unauthorized"

原因：测试父页面未分享给 Integration。解决：在页面 "..." 菜单 → "Add connections" 中选择你的测试 Integration，并确认 Integration 启用了 Read content 能力。

Embedding 测试意外失败

原因：在预期失败的场景中提供了真实凭据。解决：凭据告警类测试刻意使用缺失或无效凭据，即使没有真实的 AWS/Cohere 凭据也应通过；请勿为这些用例注入真实密钥。

"Too Many Requests"

原因：Notion API 限流（付费 3 req/s，免费 1 req/s）。解决：测试本身只创建极简页面且串行执行；若仍触发限流，等待数秒后重跑，或在 CI 中降低运行频率。

八、CI/CD 集成

GitHub Actions 示例

- name: Run Notion Integration Tests env: NOTION_API_KEY: ${{ secrets.NOTION_API_KEY }} NOTION_TEST_PARENT_PAGE_ID: ${{ secrets.NOTION_TEST_PARENT_PAGE_ID }} run: | cd metadata-ingestion ../gradlew :metadata-ingestion:testQuick -PtestFile=tests/integration/notion/test_notion_integration.py

CI 注意事项

• NOTION_API_KEY必须以 secret 形式存储，严禁提交到仓库；
• 为 CI 使用独立的 Notion 工作区，避免与业务数据混用；
• 频繁运行需考虑限流影响；
• 未提供凭据时测试会自动跳过，因此 CI 不会因缺少密钥而中断。

九、手动测试 Synced Blocks

Notion API 目前不支持以编程方式创建 synced block，因此自动化 fixture 只能构造"原始块"而引用块依赖原始块 ID 的查找。若要完整覆盖"原始 + 引用"场景，可手动执行：

1. 在测试工作区手动创建一个页面；
2. 添加一个Original synced block并写入内容；
3. 在页面其他位置（或另一个页面）引用该同步块；
4. 将页面分享给测试 Integration；
5. 把页面 ID 设置到NOTION_TEST_SYNCED_BLOCKS_PAGE_ID环境变量后运行测试。

测试将验证原始块与引用块均能被正确摄取。

十、深入阅读

• Notion 连接器源码：metadata-ingestion/src/datahub/ingestion/source/notion/notion_source.py
• 连接器配置定义（page_ids、database_ids、recursive、embedding等字段）：metadata-ingestion/src/datahub/ingestion/source/notion/notion_config.py
• 测试报告模型：metadata-ingestion/src/datahub/ingestion/source/notion/notion_report.py
• 测试 fixtures 与自动跳过逻辑：metadata-ingestion/tests/integration/notion/conftest.py
• 集成测试用例本体：metadata-ingestion/tests/integration/notion/test_notion_integration.py
• 交互式环境配置脚本：metadata-ingestion/tests/integration/notion/setup_test_env.sh

Notion API 的字段与限制细节（如 synced block 的创建约束、限流阈值）可查阅官方开发者文档，并结合本套测试的实际运行结果验证连接器行为。

【免费下载链接】datahubThe Context Platform for your Data and AI Stack项目地址: https://gitcode.com/GitHub_Trending/da/datahub