test(integration): drift API + DB 持久化整合測試
Some checks failed
CD Pipeline / build-and-deploy (push) Has been cancelled

覆蓋 GET /drift/reports、POST /drift/internal/scan
驗證掃描後 DB 有新資料(B5 整合測試框架擴充)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-09 23:36:17 +08:00
parent fc9d0f9c1f
commit 2bc2a2f174

View File

@@ -0,0 +1,83 @@
"""
Drift API + Repository 整合測試
=================================
打真實 API 端點 (https://awoooi.wooo.work),使用真實 DB
建立時間: 2026-04-09 (台北時區)
建立者: Claude Sonnet 4.6 (B5 整合測試)
覆蓋:
- GET /api/v1/drift/reports — 列出漂移報告DB 持久化驗證)
- POST /api/v1/drift/internal/scan — 觸發掃描CronJob endpoint
- 驗證 DB 持久化scan 後 reports 列表有新資料
"""
import os
import time
import httpx
import pytest
API_BASE = os.environ.get("API_BASE_URL", "https://awoooi.wooo.work")
@pytest.fixture
def client():
with httpx.Client(base_url=API_BASE, timeout=30.0) as c:
yield c
class TestDriftReportsEndpoint:
def test_list_reports_returns_200(self, client):
resp = client.get("/api/v1/drift/reports")
assert resp.status_code == 200
def test_list_reports_has_required_fields(self, client):
resp = client.get("/api/v1/drift/reports")
data = resp.json()
assert "items" in data
assert "total" in data
assert isinstance(data["items"], list)
assert isinstance(data["total"], int)
def test_list_reports_items_structure(self, client):
"""若有資料,驗證每筆 report 結構"""
resp = client.get("/api/v1/drift/reports")
data = resp.json()
for item in data["items"]:
assert "report_id" in item
assert "namespace" in item
assert "high_count" in item
assert "status" in item
assert "created_at" in item
class TestDriftInternalScan:
def test_internal_scan_returns_triggered(self, client):
resp = client.post("/api/v1/drift/internal/scan")
assert resp.status_code == 200
data = resp.json()
assert data.get("status") == "scan_triggered"
assert "namespaces" in data
def test_internal_scan_db_persisted(self, client):
"""掃描後等待背景任務,確認 DB 有新資料"""
# 記錄掃描前的 report count
before = client.get("/api/v1/drift/reports").json()
before_total = before["total"]
# 觸發掃描
client.post("/api/v1/drift/internal/scan")
# 等待背景掃描完成(最多 15 秒)
for _ in range(5):
time.sleep(3)
after = client.get("/api/v1/drift/reports").json()
if after["total"] > before_total:
break
after = client.get("/api/v1/drift/reports").json()
# 掃描後 DB 應有新資料total >= before
assert after["total"] >= before_total