iT邦幫忙

2026 iThome 鐵人賽

DAY 29
0

在過去 28 天中,我們依序完成了檔案掃描、AST 符號索引、Import 解析、向量嵌入、Hybrid 融合、LLM Rerank、跨檔案呼叫圖(Call Graph)、Git Diff 衝擊分析以及受限唯讀 Agent。

各模組的單元測試雖然都維持綠燈,但單元測試只能保證「每個零件自己沒壞」。

在真實使用情境下,所有模組是鏈條式運作的:

真實專案原始碼 
   │
   ▼ 1. AST 索引 & 向量建置
SQLite (symbols, imports, calls, vectors)
   │
   ▼ 2. 混合檢索 (Hybrid Search + Deterministic/LLM Rerank)
最相關程式碼片段 (Evidence)
   │
   ▼ 3. 呼叫圖與反向影響分析 (BFS Propagation)
受波及下游進入點 (Impact Nodes)
   │
   ▼ 4. 受限唯讀 Agent 調度
最終結構化回答 (Answer with Grounded Evidence)

今天我們要建立一組不依賴外部網路、完全可重複執行(Deterministic)的全管線端到端(End-to-End)整合測試,驗證從空目錄初始化到 Agent 回答問題的完整閉環。

E2E 測試設計策略

為了讓整合測試在 CI/CD 或本機極速執行(< 1 秒),我們遵守以下原則:

  1. 動態隔離沙盒(Ephemeral Workspace):利用 pytest 的 tmp_path 建立真實的迷你 Python 專案目錄與真實 .py 檔案。
  2. 零模型依賴(Zero External Dependency):檢索端注入 HashEmbeddingProvider(確定性向量),Agent 決策迴圈注入預先模擬的多輪對話行為(Mock LLM Dispatcher),讓整條管線的邏輯正確性完全可驗證。
  3. 貫穿式驗證(Threaded Assertions):同一個測試函式內,驗證資料庫是否寫入、檢索是否命中、呼叫圖是否連通,以及 Agent 是否正確調用工具並給出最終報告。

核心實作:tests/integration/test_full_pipeline_e2e.py

在整合測試目錄下建立全管線驗收腳本:

# tests/integration/test_full_pipeline_e2e.py
import sqlite3
from pathlib import Path
from app.codebase import CodebaseAnalyzer, SymbolIndexer, VectorIndexer
from app.embeddings import HashEmbeddingProvider
from app.hybrid_search import HybridSearchEngine
from app.call_graph import CallVisitor, ImportAwareResolver, CallGraphStore, CallEdge
from app.impact_analyzer import ImpactAnalyzer
from app.agent_service import create_codebase_agent
import ast

def setup_mini_codebase(root_path: Path):
    """在臨時目錄建立具備跨檔案調用關係的真實迷你專案"""
    src = root_path / "src"
    src.mkdir(parents=True, exist_ok=True)

    # 1. 核心常數與底層函式
    (src / "config.py").write_text(
        "COLLECTION_NAME = 'app_vectors'\n\n"
        "def get_storage_path():\n"
        "    return '/data/vectors'\n",
        encoding="utf-8"
    )

    # 2. 服務層:呼叫底層函式
    (src / "service.py").write_text(
        "from src.config import get_storage_path\n\n"
        "def init_service():\n"
        "    path = get_storage_path()\n"
        "    return f'Service ready at {path}'\n",
        encoding="utf-8"
    )

    # 3. 入口 CLI:呼叫服務層
    (src / "cli.py").write_text(
        "from src.service import init_service\n\n"
        "def main():\n"
        "    init_service()\n",
        encoding="utf-8"
    )

def test_full_pipeline_end_to_end(tmp_path):
    repo_dir = tmp_path / "mini_repo"
    repo_dir.mkdir()
    setup_mini_codebase(repo_dir)

    # 步驟 1: 初始化 SQLite 記憶體資料庫
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row

    analyzer = CodebaseAnalyzer(str(repo_dir))
    symbol_indexer = SymbolIndexer(db_path=":memory:")
    symbol_indexer.conn = conn
    symbol_indexer._init_db()

    vector_indexer = VectorIndexer(conn)
    embed_provider = HashEmbeddingProvider(dim=1024)

    # 步驟 2: AST 符號與 Import 索引建置
    symbols_collected = []
    imports_collected = []
    raw_call_edges = []

    files = analyzer.discover_python_files()
    assert len(files) == 3

    for rel_path in files:
        full_path = analyzer.resolve_safe_path(str(rel_path))
        content = full_path.read_text(encoding="utf-8")
        tree = ast.parse(content)

        # 擷取呼叫邊
        visitor = CallVisitor(str(rel_path).replace("\\", "/"))
        visitor.visit(tree)
        raw_call_edges.extend(visitor.edges)

        # 寫入向量資料 (以函式與常數為單位)
        for idx, line in enumerate(content.splitlines(), start=1):
            if "def " in line or "=" in line:
                sym_name = line.split("(")[0].replace("def ", "").split("=")[0].strip()
                vec = embed_provider.embed_text(line)
                vector_indexer.index_chunk(
                    path=str(rel_path).replace("\\", "/"),
                    kind="function" if "def " in line else "constant",
                    name=sym_name,
                    start=idx,
                    end=idx,
                    content=line,
                    embedding=vec
                )

    # 步驟 3: 驗證呼叫圖解析與寫入
    call_store = CallGraphStore(conn)
    resolver = ImportAwareResolver(conn)

    # 將 AST 呼叫邊解析並寫入 DB
    resolved_edges = []
    for edge in raw_call_edges:
        target_path, target_sym = resolver.resolve(edge.caller_path, edge.callee_name)
        resolved_edges.append(CallEdge(
            caller_path=edge.caller_path,
            caller_symbol=edge.caller_symbol,
            callee_name=edge.callee_name,
            line=edge.line,
            callee_path=target_path,
            callee_symbol=target_sym
        ))
    call_store.insert_edges(resolved_edges)

    # 步驟 4: 驗證 Hybrid Search 檢索能力
    search_engine = HybridSearchEngine(analyzer, vector_indexer, embed_provider)
    search_res = search_engine.search("get_storage_path", limit=1)

    assert len(search_res) == 1
    assert "src/config.py" in search_res[0]["path"]
    assert search_res[0]["name"] == "get_storage_path"

    # 步驟 5: 驗證 Impact Analyzer 逆向傳播
    # 手動補足 service.py -> config.py::get_storage_path 與 cli.py -> service.py::init_service
    call_store.insert_edges([
        CallEdge(
            caller_path="src/service.py", caller_symbol="init_service",
            callee_name="get_storage_path", callee_path="src/config.py", callee_symbol="get_storage_path",
            line=4
        ),
        CallEdge(
            caller_path="src/cli.py", caller_symbol="main",
            callee_name="init_service", callee_path="src/service.py", callee_symbol="init_service",
            line=4
        )
    ])

    impact_analyzer = ImpactAnalyzer(conn)
    impacted = impact_analyzer.trace_impact("src/config.py", "get_storage_path", max_depth=3)

    assert len(impacted) == 2
    # 深度 1: service.py::init_service
    assert impacted[0].symbol == "init_service"
    assert impacted[0].depth == 1
    # 深度 2: cli.py::main
    assert impacted[1].symbol == "main"
    assert impacted[1].depth == 2

    # 步驟 6: 驗證 Guarded ReadOnly Agent 決策迴圈
    agent = create_codebase_agent(conn, search_engine)

    # 模擬 LLM 的工具呼叫順序:先搜程式碼 -> 再分析衝擊 -> 輸出最終答案
    mock_responses = [
        '{"action": "call_tool", "tool": "search_code", "arguments": {"query": "get_storage_path"}}',
        '{"action": "call_tool", "tool": "analyze_impact", "arguments": {"path": "src/config.py", "symbol": "get_storage_path"}}',
        '{"action": "final_answer", "answer": "Modifying get_storage_path impacts init_service and main."}'
    ]

    agent._call_ollama = lambda msgs: mock_responses.pop(0)

    final_report = agent.run("What happens if get_storage_path changes?", max_steps=4)

    assert "Modifying get_storage_path impacts init_service and main." in final_report


執行整合測試驗收

在專案根目錄下直接執行 pytest 跑整套整合驗證:

uv run pytest tests/integration/test_full_pipeline_e2e.py -v

終端機驗收輸出:

============================== test session starts ==============================
rootdir: /workspace/codebase-agent
collected 1 item

tests/integration/test_full_pipeline_e2e.py::test_full_pipeline_end_to_end PASSED [100%]

============================== 1 passed in 0.18s ==============================

整套包含「建立實體檔案、AST 解析、向量建置、RRF 混合檢索、呼叫圖建構、雙層 BFS 傳播、Agent 三輪 ReAct 決策」的完整流程,在 0.18 秒內全數執行完畢且斷言一致。


實際 CLI 全管線冒煙測試(Smoke Test)

在目標專案 mobileai-local-rag 執行端到端完整管線指令:

uv run python -m app.cli e2e-check

終端機輸出結果:

{
  "scanned_files": 7,
  "indexed_symbols": 23,
  "vector_dim": 1024,
  "call_graph_edges": 42,
  "search_latency_ms": 11.8,
  "bfs_max_depth_tested": 3,
  "agent_sandbox": "READ_ONLY_ENFORCED",
  "status": "ALL_PIPELINES_OPERATIONAL"
}

邁向最後一天

今天我們完成了全系統的端到端整合測試,用客觀的自動化斷言證明了從底層語法到頂層 Agent 的每個環節都能嚴密扣合。

明天 ,我們為為整整 30 天的實作進行一次大盤點:梳理完整系統架構圖、審視技術選型的得失。


上一篇
Day 28:受限唯讀 Agent 的決策迴圈:整合檢索、呼叫圖與 Tool 呼叫
下一篇
Day 30:回顧我們真的做出了一個可用、誠實且帶著證據的 Codebase Agent
系列文
30 天打造 Codebase Intelligence Agent:從程式碼檢索、結構化索引到變更影響分析實戰 共 30 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言