iT邦幫忙

2026 iThome 鐵人賽

DAY 28
0
Build on Google AI

打造專屬 AI 參謀群:以 Gemini Spark、Workspace 與 ADK 構建自動化決策雷達系列 第 28 篇

Day 28:防禦型合約測試:針對 Obsidian / Notion 適配器的自動化單元測試

  • 分享至 

  • xImage
  •  

在 Day 21 到 Day 27 的高強度迭代中,我們完成了情報雷達的雙軌沉澱(Obsidian Vault 與 Notion API)、全自動無人值守推送,以及兼具 5 RPM 冷卻與模型降級的穩健防禦機制。

然而,當系統架構擴展至涵蓋多個第三方適配器(Google Sheets, Obsidian 本地檔案系統, Notion REST API, Discord Webhook)時,另一個嚴肅的軟體工程課題浮上檯面:如何保證後續重構或依賴更新時,核心契約(Contracts)不會悄悄損毀?

如果在每次測試或 CI/CD 執行單元測試時,都必須呼叫真實的 Notion API 或寫入硬碟,會面臨以下問題:

  1. 消耗外部 API 額度:浪費寶貴的 Free Tier Rate Limit(甚至直接撞 429)。
  2. 測試速度緩慢且脆弱:遭遇網路抖動或 503 時,測試就會出現隨機 Flaky 失敗。
  3. 對本地環境有副作用(Side Effects):測試執行完畢後,硬碟會多出雜亂的測試 Markdown 筆記。

今天我們將貫徹六角架構的核心優勢——透過依賴反轉(Inversion of Control)與 Mock 物件,替 ObsidianVaultAdapter 與 NotionSyncAdapter 撰寫專業的防禦型契約單元測試(Unit & Contract Tests),達成斷網環境下毫秒級 100% 綠燈通過!


一、測試架構藍圖:隔離副作用的契約驗證

六角架構的一大威力在於:業務邏輯與適配器皆遵循特定 Port 介面。測試時,我們透過 pytest 配合 unittest.mock 與內建的 tmp_path fixture,模擬外部世界的輸入與回應:

+-----------------------------------------------------------------------+
| pytest Test Suite (Zero Network / Zero Side Effects)                  |
+-----------------------------------------------------------------------+
│
├── Test Suite A: test_obsidian_adapter.py
│   ├── 使用 pytest 內建 tmp_path fixture 模擬臨時 Vault 目錄
│   ├── 驗證 Frontmatter 欄位轉譯 (YAML 格式、日期、標籤)
│   ├── 驗證檔名非法字元清洗 (Sanitization: 去除 /?*: 等)
│   └── 驗證冪等性 (檔案已存在且 overwrite=False 時安全略過)
│
└── Test Suite B: test_notion_adapter.py
    ├── 使用 unittest.mock.MagicMock 攔截 notion_client.Client
    ├── 驗證 2,000 字元長度邊界切片 (Rich Text Blocks)
    ├── 驗證 Entry ID 查詢邏輯與屬性字典映射 (Database Properties)
    └── 模擬 Notion API 404 / 429 異常時的優雅容錯回傳


二、測試環境配置

在 pyproject.toml 或 requirements-dev.txt 中加入測試依賴:

pytest>=8.0.0
pytest-mock>=3.12.0

並確認專案根目錄下有 tests/ 資料夾,結構如下:

tests/
├── __init__.py
├── adapters/
│   ├── __init__.py
│   ├── test_obsidian_adapter.py
│   └── test_notion_adapter.py
└── conftest.py


三、共用測試資料:tests/conftest.py

在 conftest.py 中建立標準的 IntelligenceItem 測試實例,供所有測試案例共用:

# tests/conftest.py
import pytest
from datetime import datetime
from src.core.models import IntelligenceItem, ItemStatus

@pytest.fixture
def sample_intelligence_item() -> IntelligenceItem:
    """提供標準、完整的已處理高價值情資作為測試基準"""
    return IntelligenceItem(
        entry_id="test_abc123",
        timestamp=datetime(2026, 10, 12, 14, 30, 0),
        status=ItemStatus.PROCESSED,
        domain="SCIENCE",
        track="AI_Frontier",
        source_title="Dynamic State Locking: Novel Approach for Agentic Systems",
        source_url="https://arxiv.org/abs/2610.99999",
        raw_content="針對分散式多代理人系統,提出動態狀態鎖定協議,保證並發情境下狀態機的一致性。",
        metrics="吞吐量提升 45%,鎖定衝突率降低至 0.02%",
        limitations="跨節點網路延遲大於 200ms 時會降級為悲觀鎖",
        raw_score=5,
        verified_score=5,
        editorial_summary="技術審查:確認此架構兼具理論與實務價值,有效避免分散式管線重入衝突。",
        cross_impact_notes="此架構可直接搬移至金融高頻訂單簿與工業機器人協同調度。",
        metadata={"reviewer_decision": "PASS"}
    )


四、測試實作 A:ObsidianVaultAdapter 單元測試

測試重點聚焦在:

  1. 檔案系統安全性:特殊字元(如 :、/、?)是否被正確清洗。
  2. Frontmatter 格式:YAML 標頭與 Obsidian Callout 標籤是否正確渲染。
  3. 冪等性防護:重複同步是否正確返回 False 並略過。
# tests/adapters/test_obsidian_adapter.py
from pathlib import Path
from src.adapters.obsidian_adapter import ObsidianVaultAdapter
from src.core.models import IntelligenceItem

def test_obsidian_sync_success(tmp_path: Path, sample_intelligence_item: IntelligenceItem):
    """驗證筆記檔案能夠正確生成,且 YAML Frontmatter 內容符合規格"""
    adapter = ObsidianVaultAdapter(vault_path=tmp_path, subfolder="Radar_Inbox")
    
    # 執行同步
    success = adapter.sync_item(sample_intelligence_item, overwrite=False)
    
    assert success is True
    
    # 檢查目標檔案是否存在
    expected_files = list((tmp_path / "Radar_Inbox").glob("*.md"))
    assert len(expected_files) == 1
    
    note_path = expected_files[0]
    content = note_path.read_text(encoding="utf-8")
    
    # 驗證核心屬性與 Markdown 結構
    assert 'id: "test_abc123"' in content
    assert 'domain: "SCIENCE"' in content
    assert 'score: 5' in content
    assert "> [!abstract] 核心提煉" in content
    assert "動態狀態鎖定協議" in content
    assert "[[SCIENCE]]" in content

def test_obsidian_filename_sanitization(tmp_path: Path, sample_intelligence_item: IntelligenceItem):
    """驗證標題包含作業系統非法字元時,能被安全清洗且不引發異常"""
    sample_intelligence_item.source_title = 'What is "AGI"?: A/B Testing & Scaling*Rules'
    adapter = ObsidianVaultAdapter(vault_path=tmp_path, subfolder="Radar_Inbox")
    
    success = adapter.sync_item(sample_intelligence_item)
    assert success is True
    
    generated_file = list((tmp_path / "Radar_Inbox").glob("*.md"))[0]
    filename = generated_file.name
    
    # 確認非法字元已全數拔除
    for char in ['"', '?', ':', '/', '*', '<', '>', '|']:
        assert char not in filename

def test_obsidian_idempotency_skip(tmp_path: Path, sample_intelligence_item: IntelligenceItem):
    """驗證冪等性:當檔案已存在且未指定 overwrite 時,應優雅略過並回傳 False"""
    adapter = ObsidianVaultAdapter(vault_path=tmp_path, subfolder="Radar_Inbox")
    
    # 首次寫入
    first_run = adapter.sync_item(sample_intelligence_item, overwrite=False)
    assert first_run is True
    
    # 第二次重複寫入(不覆寫)
    second_run = adapter.sync_item(sample_intelligence_item, overwrite=False)
    assert second_run is False


五、測試實作 B:NotionSyncAdapter 契約與 Mock 測試

測試重點聚焦在:

  1. API 呼叫契約:是否向 Notion REST API 送出正確的 JSON 結構(parent、properties、children)。
  2. 查詢防重複:已存在該 Entry ID 時是否觸發 pages.update 而非 pages.create。
  3. 例外防護:當 Notion API 回傳例外時,是否安全記錄並返回 False 而不崩潰。
# tests/adapters/test_notion_adapter.py
from unittest.mock import MagicMock, patch
from notion_client.errors import APIResponseError
from httpx import Response, Request
from src.adapters.notion_adapter import NotionSyncAdapter
from src.core.models import IntelligenceItem

@patch("src.adapters.notion_adapter.Client")
def test_notion_create_new_page(mock_client_class, sample_intelligence_item: IntelligenceItem):
    """驗證情資首次同步時,正確呼叫 pages.create 建立結構化頁面"""
    mock_client = MagicMock()
    mock_client_class.return_value = mock_client
    
    # 模擬資料庫查詢回傳空清單 (表示不存在此 entry_id)
    mock_client.databases.query.return_value = {"results": []}
    
    adapter = NotionSyncAdapter(auth_token="fake_token", database_id="fake_db_123")
    result = adapter.sync_item(sample_intelligence_item)
    
    assert result is True
    mock_client.databases.query.assert_called_once()
    
    # 驗證 pages.create 被呼叫,且參數正確封裝
    mock_client.pages.create.assert_called_once()
    call_args = mock_client.pages.create.call_args[1]
    
    assert call_args["parent"] == {"database_id": "fake_db_123"}
    props = call_args["properties"]
    assert props["Title"]["title"][0]["text"]["content"] == sample_intelligence_item.source_title
    assert props["Domain"]["select"]["name"] == "SCIENCE"
    assert props["Score"]["number"] == 5.0

@patch("src.adapters.notion_adapter.Client")
def test_notion_update_existing_page(mock_client_class, sample_intelligence_item: IntelligenceItem):
    """驗證情資已存在且 overwrite=True 時,正確呼叫 pages.update 進行覆寫"""
    mock_client = MagicMock()
    mock_client_class.return_value = mock_client
    
    # 模擬資料庫中已有一筆相同 Entry ID 的頁面
    mock_client.databases.query.return_value = {
        "results": [{"id": "existing_notion_page_uuid"}]
    }
    
    adapter = NotionSyncAdapter(auth_token="fake_token", database_id="fake_db_123")
    
    # 測試 overwrite=False (應略過)
    skip_result = adapter.sync_item(sample_intelligence_item, overwrite=False)
    assert skip_result is False
    mock_client.pages.update.assert_not_called()
    
    # 測試 overwrite=True (應更新既有頁面)
    update_result = adapter.sync_item(sample_intelligence_item, overwrite=True)
    assert update_result is True
    mock_client.pages.update.assert_called_once()
    assert mock_client.pages.update.call_args[1]["page_id"] == "existing_notion_page_uuid"

@patch("src.adapters.notion_adapter.Client")
def test_notion_api_error_handling(mock_client_class, sample_intelligence_item: IntelligenceItem):
    """驗證當 Notion API 拋出連線或驗證錯誤時,適配器不會向上崩潰,而是回傳 False"""
    mock_client = MagicMock()
    mock_client_class.return_value = mock_client
    
    # 模擬查詢時發生 HTTP 500 異常
    dummy_req = Request("POST", "https://api.notion.com")
    dummy_res = Response(500, request=dummy_req)
    mock_client.databases.query.side_effect = APIResponseError(dummy_res, "Internal Server Error", "server_error")
    
    adapter = NotionSyncAdapter(auth_token="fake_token", database_id="fake_db_123")
    result = adapter.sync_item(sample_intelligence_item)
    
    # 斷言:適配器內部捕獲錯誤,回傳 False 保證管線不拋錨
    assert result is False


六、執行測試與 CI 整合

在本地終端機執行測試套件:

pytest -v tests/adapters/

執行輸出:

============================= test session starts ==============================
platform darwin -- Python 3.11.x, pytest-8.3.2
rootdir: /Users/mervyn/Projects/intelligence-radar
collected 6 items

tests/adapters/test_obsidian_adapter.py::test_obsidian_sync_success PASSED   [ 16%]
tests/adapters/test_obsidian_adapter.py::test_obsidian_filename_sanitization PASSED [ 33%]
tests/adapters/test_obsidian_adapter.py::test_obsidian_idempotency_skip PASSED [ 50%]
tests/adapters/test_notion_adapter.py::test_notion_create_new_page PASSED     [ 66%]
tests/adapters/test_notion_adapter.py::test_notion_update_existing_page PASSED [ 83%]
tests/adapters/test_notion_adapter.py::test_notion_api_error_handling PASSED [100%]

============================== 6 passed in 0.12s ===============================

整套測試在完全不連外網、不花一分錢 Token、不產生任何硬碟殘留垃圾的狀態下,僅需 0.12 秒 便全部綠燈通過!

GitHub Actions CI 自動化把關

在 .github/workflows/patrol.yml 中,在執行正式巡檢前加入單元測試步驟:

      - name: Run Contract & Unit Tests
        run: |
          pytest tests/ -q

這確保了在每次程式碼 Push 或 Pull Request 時,只要有人不小心改壞了適配器格式,CI 流程會在第一時間攔截阻斷,防止瑕疵邏輯流入生產環境。


總結與下一步

今天我們為多轉接器架構加上了最堅固的護城河:

  • 貫徹了六角架構可測試性(Testability)的核心價值,透過 pytest、tmp_path 與 MagicMock 徹底隔離副作用。
  • 驗證了 Obsidian 適配器的合法字元清洗、Frontmatter 渲染與冪等略過邏輯。
  • 驗證了 Notion 適配器在新增、更新與 API 例外下的契約正確性。

隨著系統在功能、防禦、可觀測性與測試層面全面達標,我們即將抵達 30 天馬拉松的終點線。

在 Day 29,我們將抽離具體代碼,站在更高維度的架構視角進行深度覆盤:「架構復盤:六角架構在自主智能體(Agentic Systems)中的真實威力」,探討 Ports & Adapters 如何讓 LLM 系統在快速迭代的時代中始終優雅敏捷!


上一篇
Day 27:重試、熔斷與配額防禦:打造穩健的 Exponential Backoff 與降級機制
下一篇
Day 29:架構復盤:六角架構在自主智能體(Agentic Systems)中的真實威力
系列文
打造專屬 AI 參謀群:以 Gemini Spark、Workspace 與 ADK 構建自動化決策雷達 共 29 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言