在 Day 21 到 Day 27 的高強度迭代中,我們完成了情報雷達的雙軌沉澱(Obsidian Vault 與 Notion API)、全自動無人值守推送,以及兼具 5 RPM 冷卻與模型降級的穩健防禦機制。
然而,當系統架構擴展至涵蓋多個第三方適配器(Google Sheets, Obsidian 本地檔案系統, Notion REST API, Discord Webhook)時,另一個嚴肅的軟體工程課題浮上檯面:如何保證後續重構或依賴更新時,核心契約(Contracts)不會悄悄損毀?
如果在每次測試或 CI/CD 執行單元測試時,都必須呼叫真實的 Notion API 或寫入硬碟,會面臨以下問題:
今天我們將貫徹六角架構的核心優勢——透過依賴反轉(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"}
)
ObsidianVaultAdapter 單元測試測試重點聚焦在:
:、/、?)是否被正確清洗。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
NotionSyncAdapter 契約與 Mock 測試測試重點聚焦在:
parent、properties、children)。Entry ID 時是否觸發 pages.update 而非 pages.create。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
在本地終端機執行測試套件:
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/workflows/patrol.yml 中,在執行正式巡檢前加入單元測試步驟:
- name: Run Contract & Unit Tests
run: |
pytest tests/ -q
這確保了在每次程式碼 Push 或 Pull Request 時,只要有人不小心改壞了適配器格式,CI 流程會在第一時間攔截阻斷,防止瑕疵邏輯流入生產環境。
今天我們為多轉接器架構加上了最堅固的護城河:
pytest、tmp_path 與 MagicMock 徹底隔離副作用。隨著系統在功能、防禦、可觀測性與測試層面全面達標,我們即將抵達 30 天馬拉松的終點線。
在 Day 29,我們將抽離具體代碼,站在更高維度的架構視角進行深度覆盤:「架構復盤:六角架構在自主智能體(Agentic Systems)中的真實威力」,探討 Ports & Adapters 如何讓 LLM 系統在快速迭代的時代中始終優雅敏捷!