Day 28：防禦性同步測試！為 Notion 與 Obsidian 打造合約測試與冪等性安全網（含實戰避坑指南）

在前面的章節中，我們實現了雙知識庫同步適配器（NotionSyncAdapter 與 ObsidianVaultAdapter），讓經由四大 Agent 提煉過的高價值情資能夠自動分流沈澱到 Notion 資料庫與本地 Obsidian Vault 中。

然而，當系統進入長期無人值守的自動化巡檢時，最棘手的問題往往出在「外部服務的副作用與非預期改動」：

• 每次改動程式碼都要真實打一次 Notion API 嗎？ 這不僅速度極慢，還可能因超出 Rate Limit、網路波動造成測試不穩定，甚至污染雲端資料庫。
• 改動了 Jinja2 模板，會不會漏掉了重要欄位？
• 重複執行時，系統是否能正確觸發「冪等性略過」，避免重複新增或覆蓋既有筆記？

今天我們將透過 pytest 建立合約測試（Contract Testing）與隔離 Mock 機制，在毫秒級內驗證轉接器的冪等性與異常防護，並為 CI/CD 巡檢管線築起最堅固的質量防線！

一、Day 28 的核心目的：為什麼要寫這些測試？

本章節的測試絕不僅是追求測試覆蓋率的「綠燈數字」，而是為系統架構注入以下三大核心防禦能力：

1. 六角架構的「合約對齊（Contract Verification）」：
   我們透過領域介面 KnowledgeSyncPort 規範了同步協定。單元測試的目的在於確保無論抽換成 Notion 還是 Obsidian，任何適配器都必須嚴格遵守相同的資料合約（正確處理 IntelligenceItem 的 15 個標準欄位）。

2. 純記憶體隔離與零副作用（Zero-Side-Effect Mocking）：
   透過 unittest.mock 模擬 Notion API 的網路請求與回應，並使用 pytest 內建的 tmp_path 模擬本地磁碟目錄，完全不依賴真實網路與外部金鑰即可完成完整驗證。

3. 冪等性（Idempotency）守護網：
   驗證適配器在遭遇重複情資時，在預設 overwrite=False 下能否精準略過並回傳 False，防止資料重複堆疊或無謂覆蓋。

二、全域測試基石：標準化 Fixture 設計 (tests/conftest.py)

為了避免在多個測試檔案中重複建立不完整的假資料，導致 Pydantic 模型驗證失敗，我們在 tests/conftest.py 中建立符合全域合約的 sample_intelligence_item：

# tests/conftest.py
from datetime import datetime, timezone
import pytest
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, tzinfo=timezone.utc),
        status=ItemStatus.PROCESSED,
        domain="SCIENCE",
        track="AI_Frontier",
        source_title="動態狀態鎖定協議：分散式代理人架構實作",
        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"}
    )

重點提示：Pydantic 領域模型中的 track 與 raw_score 為必填欄位，且 entry_id 與核心文字（如「動態狀態鎖定協議」）必須完整貫穿標題與摘要，以供後續模板斷言比對。

三、適配器測試實作

1. Notion 適配器測試 (tests/adapters/test_notion_adapter.py)

Notion 測試的核心在於模擬 API 互動、驗證新頁面建立、覆寫判斷以及 API 異常攔截：

# tests/adapters/test_notion_adapter.py
from unittest.mock import MagicMock, patch
import pytest
from notion_client.errors import APIResponseError
from src.adapters.notion_adapter import NotionSyncAdapter

# 使用合法的 32 碼十六進位 UUID 避免初始化校驗失敗
MOCK_DB_ID = "3f44e1c203bf8097815fef01e424ddc1"

def test_notion_create_new_page(sample_intelligence_item):
    adapter = NotionSyncAdapter(auth_token="fake_token", database_id=MOCK_DB_ID)
    
    # 模擬該條目尚未存在於 Notion 中
    with patch.object(adapter, "_find_page_by_entry_id", return_value=None):
        with patch.object(adapter.client.pages, "create") as mock_create:
            mock_create.return_value = {"id": "new_page_999"}
            success = adapter.sync_item(sample_intelligence_item)
            assert success is True
            mock_create.assert_called_once()

def test_notion_update_existing_page(sample_intelligence_item):
    adapter = NotionSyncAdapter(auth_token="fake_token", database_id=MOCK_DB_ID)
    
    # 模擬該條目已存在
    with patch.object(adapter, "_find_page_by_entry_id", return_value="existing_page_888"):
        # 1. 預設 overwrite=False：必須安全略過 (冪等性保障)
        success_skip = adapter.sync_item(sample_intelligence_item, overwrite=False)
        assert success_skip is False
        
        # 2. 指定 overwrite=True：觸發更新
        with patch.object(adapter.client.pages, "update") as mock_update:
            mock_update.return_value = {"id": "existing_page_888"}
            success_update = adapter.sync_item(sample_intelligence_item, overwrite=True)
            assert success_update is True
            mock_update.assert_called_once()

def test_notion_api_error_handling(sample_intelligence_item):
    """驗證當 Notion API 拋出異常時，轉接器能安全捕獲並回傳 False，不造成程式崩潰"""
    adapter = NotionSyncAdapter(auth_token="fake_token", database_id=MOCK_DB_ID)
    
    # 動態建構 APIResponseError 實例，繞過不同版本底層 __init__ 引數差異
    mock_err = APIResponseError.__new__(APIResponseError)
    Exception.__init__(mock_err, "Mock Notion API 400 Bad Request")
    mock_err.code = "validation_error"
    mock_err.status = 400

    with patch.object(adapter, "_find_page_by_entry_id", return_value=None):
        with patch.object(adapter.client.pages, "create", side_effect=mock_err):
            success = adapter.sync_item(sample_intelligence_item)
            assert success is False

2. Obsidian 適配器測試 (tests/adapters/test_obsidian_adapter.py)

Obsidian 測試利用 pytest 的 tmp_path 建立虛擬沙盒，驗證 Markdown 模板渲染、檔名字元安全過濾與磁碟覆寫控制：

# 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):
    adapter = ObsidianVaultAdapter(vault_path=tmp_path)
    
    success = adapter.sync_item(sample_intelligence_item)
    assert success is True

    # 檢查目標檔案是否確實產生
    target_files = list((tmp_path / "00_Inbox/Radar").glob("*.md"))
    assert len(target_files) == 1

    content = target_files[0].read_text(encoding="utf-8")
    # 斷言關鍵合約欄位與摘要是否成功渲染
    assert f'id: "{sample_intelligence_item.entry_id}"' in content
    assert "動態狀態鎖定協議" in content

def test_obsidian_idempotency_skip(tmp_path: Path, sample_intelligence_item: IntelligenceItem):
    adapter = ObsidianVaultAdapter(vault_path=tmp_path)
    
    # 第一次寫入成功
    assert adapter.sync_item(sample_intelligence_item) is True
    # 第二次重複寫入（未開啟 overwrite）必須略過並回傳 False
    assert adapter.sync_item(sample_intelligence_item, overwrite=False) is False

def test_obsidian_filename_sanitization(tmp_path: Path, sample_intelligence_item: IntelligenceItem):
    adapter = ObsidianVaultAdapter(vault_path=tmp_path)
    
    # 注入含有非法路徑字元（冒號、斜線、問號）的標題
    sample_intelligence_item.source_title = "Breaking/Alert: Are Agents Ready? [v1.0]"
    
    assert adapter.sync_item(sample_intelligence_item) is True
    target_files = list((tmp_path / "00_Inbox/Radar").glob("*.md"))
    # 驗證檔名未包含任何非法符號
    filename = target_files[0].name
    for char in ['/', '\\', ':', '*', '?', '"', '<', '>', '|']:
        assert char not in filename

四、在 CI/CD 中建立安全門禁 (patrol.yml)

有了這套合約測試後，我們在 .github/workflows/patrol.yml 中將測試安置在關鍵的門禁節點：

• 執行時機：在環境準備與金鑰解碼後，正式執行 run_pipeline.py 前！
• 門禁效益：只要任何單元測試失敗，流程立即熔斷（Fail-Fast），絕不浪費 API 額度或對正式試算表進行誤操作。

      # 解碼 GCP 金鑰後，立即執行合約與單元測試
      - name: Run Contract & Unit Tests
        env:
          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
          NOTION_API_TOKEN: ${{ secrets.NOTION_API_TOKEN }}
          NOTION_DATABASE_ID: ${{ secrets.NOTION_DATABASE_ID }}
        run: |
          pytest -v
          
      # 測試全數通過後，才啟動正式情報流水線
      - name: Execute Intelligence Pipeline
        run: |
          python run_pipeline.py

五、💡 實戰避坑指南 (Troubleshooting FAQ)

Q1: 測試時拋出 ValueError: badly formed hexadecimal UUID string？

• 原因：為了嚴格防止 Notion API 404/400 錯誤，適配器在初始化時會以 uuid.UUID 校驗 database_id。如果單元測試傳入 database_id="fake_db_123" 等非 32 碼十六進位字串，就會在初始化時中斷。
• 解法：在測試中使用合法的 32 碼假 UUID（如 "3f44e1c203bf8097815fef01e424ddc1"），並在適配器建構式中加入 try-except ValueError 進行測試容錯相容。

Q2: 建立 APIResponseError 時拋出 TypeError: HTTPResponseError.__init__() missing required positional argument？

• 原因：notion-client 底層的 APIResponseError 繼承自 HTTPResponseError，其建構函式因版本不同會要求 status、message、headers、raw_body_text 等多個位置引數。
• 解法：在單元測試中，透過 APIResponseError.__new__(APIResponseError) 動態生成實例，直接掛載 status 與 code 屬性，徹底擺脫不同版本間引數簽章不相容的困擾。

Q3: test_obsidian_sync_success 斷言失敗（找不到特定關鍵字）？

• 原因：通常是資料模型欄位與 Jinja2 模板命名脫節。例如資料模型中使用 cross_impact_notes，而 Markdown 模板卻寫成 {{ item.second_order_effects }}；或測試關鍵字只填在 raw_content，但模板根本沒有渲染原始內容。
• 解法：在 Jinja2 模板中採用相容寫法 {{ item.cross_impact_notes or item.second_order_effects }}，並確保測試 fixture 的標題與摘要欄位均包含斷言所需的目標字串。

Q4: tests/ 資料夾底下的程式碼需要提交到 Git 嗎？

• 說明：一定要 commit 到 Git 中！ 雲端 GitHub Actions 執行 pytest 時必須依賴儲存庫中的 tests/ 目錄。只有測試自動生成的暫存檔案（如 .pytest_cache/、.coverage、htmlcov/）才需要列入 .gitignore。

六、結語

透過合約測試，我們為系統在對接第三方服務（Notion）與本地儲存（Obsidian）時建立了一道不可跨越的安全邊界：

1. 重構不再提心吊膽：未來不論如何調整 Markdown 模板結構或重構 API 連線，只要執行 pytest，一秒之內即可確認契約是否完好。
2. 阻絕雲端意外：確保每一次自動化巡邏都有最扎實的代碼品質防護，實踐真正穩健的自動化情報雷達！