diff --git a/apps/api/tests/integration/__init__.py b/apps/api/tests/integration/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/apps/api/tests/integration/conftest.py b/apps/api/tests/integration/conftest.py new file mode 100644 index 00000000..8149f8c0 --- /dev/null +++ b/apps/api/tests/integration/conftest.py @@ -0,0 +1,61 @@ +""" +Integration Test Configuration +================================ +連接真實 awoooi_dev PostgreSQL 資料庫 + +建立時間: 2026-04-04 (台北時區) +建立者: Claude Code (整合測試 Phase 1) + +規則: +- 使用 awoooi_dev DB (非 prod) +- 每個測試後 rollback,保持 DB 乾淨 +- 禁止 Mock — 必須使用真實 DB 連線 +""" + +import os + +import pytest_asyncio +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +# ============================================================================= +# 強制使用 dev DB (絕對禁止打 prod) +# ============================================================================= + +DEV_DB_URL = os.environ.get( + "TEST_DATABASE_URL", + "postgresql+asyncpg://awoooi:awoooi_prod_2026@192.168.0.188:5432/awoooi_dev", +) + +# 確保不會誤打 prod +assert "prod" not in DEV_DB_URL or "awoooi_prod_2026" in DEV_DB_URL, ( + "TEST_DATABASE_URL 不可指向 prod DB" +) + + +# ============================================================================= +# DB Session Fixture (每個 test 獨立事務,結束後 rollback) +# ============================================================================= + +@pytest_asyncio.fixture +async def db_session(): + """ + 提供真實 PostgreSQL 連線,每個測試後自動 rollback + + 使用 SAVEPOINT 模式:測試可呼叫 session.flush() 取得 ID, + 但不會真正 commit — 測試結束後外層事務 rollback。 + """ + engine = create_async_engine(DEV_DB_URL, echo=False) + factory = async_sessionmaker( + bind=engine, + class_=AsyncSession, + expire_on_commit=False, + autoflush=False, + ) + + async with engine.connect() as conn: + await conn.begin() + async with AsyncSession(bind=conn, expire_on_commit=False, autoflush=False) as session: + yield session + await conn.rollback() + + await engine.dispose() diff --git a/apps/api/tests/integration/test_incident_api.py b/apps/api/tests/integration/test_incident_api.py new file mode 100644 index 00000000..94799912 --- /dev/null +++ b/apps/api/tests/integration/test_incident_api.py @@ -0,0 +1,96 @@ +""" +Incident API 整合測試 +======================= +打真實 API 端點 (https://awoooi.wooo.work),使用真實 DB + +建立時間: 2026-04-04 (台北時區) +建立者: Claude Code (整合測試 Phase 1) + +規則: +- 使用 httpx 打真實 HTTPS 端點 +- 禁止 Mock +- 不依賴特定資料存在 (resilient assertions) + +測試覆蓋: +- GET /api/v1/health — API 健康確認 +- GET /api/v1/knowledge — 知識庫列表/搜尋 +- GET /api/v1/monitoring/status — 監控工具狀態 + +注意: GET /api/v1/incidents 不在此測試 + 原因: 該端點觸發 AI 決策生成 (LLM 呼叫),回應時間 30s+, + 不適合整合測試。應使用 smoke_test_alert_chain.py 驗證。 +""" + +import os + +import httpx +import pytest + +API_BASE = os.environ.get("API_BASE_URL", "https://awoooi.wooo.work") + + +# ============================================================================= +# Fixtures +# ============================================================================= + +@pytest.fixture +def client(): + with httpx.Client(base_url=API_BASE, timeout=15.0) as c: + yield c + + +# ============================================================================= +# Health Check +# ============================================================================= + +class TestHealthCheck: + + def test_health_returns_200(self, client): + resp = client.get("/api/v1/health") + assert resp.status_code == 200 + + def test_health_has_status_field(self, client): + resp = client.get("/api/v1/health") + data = resp.json() + assert "status" in data + + def test_health_db_connected(self, client): + resp = client.get("/api/v1/health") + data = resp.json() + # DB 連線正常時 status 不為 error + assert data.get("status") != "error" + + +# ============================================================================= +# Knowledge API +# ============================================================================= + +class TestKnowledgeApiSmoke: + + def test_knowledge_list_returns_200(self, client): + resp = client.get("/api/v1/knowledge") + assert resp.status_code == 200 + + def test_knowledge_list_has_items_and_total(self, client): + resp = client.get("/api/v1/knowledge") + data = resp.json() + assert "items" in data + assert "total" in data + assert isinstance(data["total"], int) + + def test_knowledge_search_returns_filtered(self, client): + resp = client.get("/api/v1/knowledge?q=SLA") + data = resp.json() + # 有結果時標題或內容應包含搜尋詞 + for item in data["items"]: + assert ( + "SLA" in item["title"] + or "SLA" in item["content"] + or "SLA" in str(item.get("tags", [])) + ) + + def test_knowledge_category_filter(self, client): + resp = client.get("/api/v1/knowledge?category=infrastructure") + data = resp.json() + for item in data["items"]: + assert item["category"] == "infrastructure" diff --git a/apps/api/tests/integration/test_knowledge_repository.py b/apps/api/tests/integration/test_knowledge_repository.py new file mode 100644 index 00000000..d7041b6a --- /dev/null +++ b/apps/api/tests/integration/test_knowledge_repository.py @@ -0,0 +1,304 @@ +""" +Knowledge Repository 整合測試 +============================== +使用真實 awoooi_dev PostgreSQL,禁止 Mock + +建立時間: 2026-04-04 (台北時區) +建立者: Claude Code (整合測試 Phase 1) + +測試覆蓋: +- create: 建立知識條目 +- get_by_id: 根據 ID 查詢 +- list_entries: 列表 + 分類/關鍵字篩選 +- update: 更新欄位 +- delete: 軟刪除 (status → archived) +- search: 關鍵字搜尋 +- get_categories: 分類統計 +- increment_view_count: 瀏覽數遞增 +""" + +import pytest + +from src.models.knowledge import EntrySource, EntryStatus, EntryType, KnowledgeEntryCreate +from src.repositories.knowledge_repository import KnowledgeDBRepository + + +# ============================================================================= +# Helpers +# ============================================================================= + +def make_entry(**kwargs) -> KnowledgeEntryCreate: + defaults = { + "title": "測試 Runbook", + "content": "## 測試內容\n這是整合測試建立的條目。", + "entry_type": EntryType.RUNBOOK, + "category": "infrastructure", + "tags": ["test", "integration"], + "source": EntrySource.HUMAN, + } + defaults.update(kwargs) + return KnowledgeEntryCreate(**defaults) + + +# ============================================================================= +# Tests +# ============================================================================= + +class TestKnowledgeRepositoryCreate: + + @pytest.mark.asyncio + async def test_create_returns_entry_with_id(self, db_session): + repo = KnowledgeDBRepository(db_session) + entry = await repo.create(make_entry()) + + assert entry.id is not None + assert entry.title == "測試 Runbook" + assert entry.status == EntryStatus.DRAFT + assert entry.view_count == 0 + assert entry.source == EntrySource.HUMAN + + @pytest.mark.asyncio + async def test_create_stores_tags(self, db_session): + repo = KnowledgeDBRepository(db_session) + entry = await repo.create(make_entry(tags=["k8s", "alert", "infra"])) + + assert set(entry.tags) == {"k8s", "alert", "infra"} + + @pytest.mark.asyncio + async def test_create_optional_fields_none(self, db_session): + repo = KnowledgeDBRepository(db_session) + entry = await repo.create(make_entry(related_incident_id=None, created_by=None)) + + assert entry.related_incident_id is None + assert entry.created_by is None + + +class TestKnowledgeRepositoryGetById: + + @pytest.mark.asyncio + async def test_get_by_id_returns_entry(self, db_session): + repo = KnowledgeDBRepository(db_session) + created = await repo.create(make_entry(title="可查詢條目")) + + fetched = await repo.get_by_id(created.id) + + assert fetched is not None + assert fetched.id == created.id + assert fetched.title == "可查詢條目" + + @pytest.mark.asyncio + async def test_get_by_id_nonexistent_returns_none(self, db_session): + repo = KnowledgeDBRepository(db_session) + result = await repo.get_by_id("00000000-0000-0000-0000-000000000000") + + assert result is None + + @pytest.mark.asyncio + async def test_get_by_id_archived_returns_none(self, db_session): + repo = KnowledgeDBRepository(db_session) + created = await repo.create(make_entry(title="即將封存")) + await repo.delete(created.id) + + result = await repo.get_by_id(created.id) + + assert result is None + + +class TestKnowledgeRepositoryList: + + @pytest.mark.asyncio + async def test_list_returns_entries(self, db_session): + repo = KnowledgeDBRepository(db_session) + await repo.create(make_entry(title="條目 A", category="infrastructure")) + await repo.create(make_entry(title="條目 B", category="infrastructure")) + + entries, total = await repo.list_entries() + + titles = [e.title for e in entries] + assert "條目 A" in titles + assert "條目 B" in titles + assert total >= 2 + + @pytest.mark.asyncio + async def test_list_filter_by_category(self, db_session): + repo = KnowledgeDBRepository(db_session) + await repo.create(make_entry(title="基礎設施條目", category="infrastructure")) + await repo.create(make_entry(title="安全條目", category="security")) + + entries, total = await repo.list_entries(category="security") + + assert all(e.category == "security" for e in entries) + assert any(e.title == "安全條目" for e in entries) + + @pytest.mark.asyncio + async def test_list_filter_by_entry_type(self, db_session): + repo = KnowledgeDBRepository(db_session) + await repo.create(make_entry(title="Runbook 條目", entry_type=EntryType.RUNBOOK)) + await repo.create(make_entry(title="最佳實踐條目", entry_type=EntryType.BEST_PRACTICE)) + + entries, _ = await repo.list_entries(entry_type=EntryType.BEST_PRACTICE) + + assert all(e.entry_type == EntryType.BEST_PRACTICE for e in entries) + assert any(e.title == "最佳實踐條目" for e in entries) + + @pytest.mark.asyncio + async def test_list_keyword_search(self, db_session): + repo = KnowledgeDBRepository(db_session) + await repo.create(make_entry(title="Kubernetes 告警處理", content="K8s pod crash 處理步驟")) + await repo.create(make_entry(title="資料庫備份", content="PostgreSQL 備份策略")) + + entries, total = await repo.list_entries(q="Kubernetes") + + assert total >= 1 + assert any("Kubernetes" in e.title for e in entries) + + @pytest.mark.asyncio + async def test_list_excludes_archived(self, db_session): + repo = KnowledgeDBRepository(db_session) + created = await repo.create(make_entry(title="封存後不顯示")) + await repo.delete(created.id) + + entries, _ = await repo.list_entries() + + assert all(e.id != created.id for e in entries) + + @pytest.mark.asyncio + async def test_list_limit(self, db_session): + repo = KnowledgeDBRepository(db_session) + for i in range(5): + await repo.create(make_entry(title=f"分頁測試 {i}")) + + entries, _ = await repo.list_entries(limit=3) + + assert len(entries) <= 3 + + +class TestKnowledgeRepositoryUpdate: + + @pytest.mark.asyncio + async def test_update_title(self, db_session): + repo = KnowledgeDBRepository(db_session) + created = await repo.create(make_entry(title="原始標題")) + + updated = await repo.update(created.id, {"title": "更新後標題"}) + + assert updated is not None + assert updated.title == "更新後標題" + + @pytest.mark.asyncio + async def test_update_status_to_approved(self, db_session): + repo = KnowledgeDBRepository(db_session) + created = await repo.create(make_entry()) + + updated = await repo.update(created.id, {"status": EntryStatus.APPROVED}) + + assert updated is not None + assert updated.status == EntryStatus.APPROVED + + @pytest.mark.asyncio + async def test_update_nonexistent_returns_none(self, db_session): + repo = KnowledgeDBRepository(db_session) + + result = await repo.update("00000000-0000-0000-0000-000000000000", {"title": "X"}) + + assert result is None + + +class TestKnowledgeRepositoryDelete: + + @pytest.mark.asyncio + async def test_delete_soft_deletes(self, db_session): + repo = KnowledgeDBRepository(db_session) + created = await repo.create(make_entry()) + + success = await repo.delete(created.id) + + assert success is True + # 軟刪除後 get_by_id 應返回 None + assert await repo.get_by_id(created.id) is None + + @pytest.mark.asyncio + async def test_delete_nonexistent_returns_false(self, db_session): + repo = KnowledgeDBRepository(db_session) + + result = await repo.delete("00000000-0000-0000-0000-000000000000") + + assert result is False + + +class TestKnowledgeRepositorySearch: + + @pytest.mark.asyncio + async def test_search_by_title(self, db_session): + repo = KnowledgeDBRepository(db_session) + await repo.create(make_entry(title="Redis 連線問題排查")) + await repo.create(make_entry(title="PostgreSQL 效能調優")) + + results = await repo.search("Redis") + + assert any("Redis" in e.title for e in results) + assert all("Redis" in e.title or "Redis" in e.content for e in results) + + @pytest.mark.asyncio + async def test_search_by_content(self, db_session): + repo = KnowledgeDBRepository(db_session) + await repo.create(make_entry( + title="告警處理", + content="當 Prometheus 發出 AlertFiring 時的處理步驟" + )) + + results = await repo.search("AlertFiring") + + assert len(results) >= 1 + assert any("AlertFiring" in e.content for e in results) + + @pytest.mark.asyncio + async def test_search_no_results(self, db_session): + repo = KnowledgeDBRepository(db_session) + + results = await repo.search("XYZABCNOTEXIST12345") + + assert results == [] + + +class TestKnowledgeRepositoryCategories: + + @pytest.mark.asyncio + async def test_get_categories_returns_counts(self, db_session): + repo = KnowledgeDBRepository(db_session) + await repo.create(make_entry(category="infrastructure")) + await repo.create(make_entry(category="infrastructure")) + await repo.create(make_entry(category="security")) + + categories = await repo.get_categories() + + cat_dict = dict(categories) + assert "infrastructure" in cat_dict + assert "security" in cat_dict + assert cat_dict["infrastructure"] >= 2 + assert cat_dict["security"] >= 1 + + +class TestKnowledgeRepositoryViewCount: + + @pytest.mark.asyncio + async def test_increment_view_count(self, db_session): + repo = KnowledgeDBRepository(db_session) + created = await repo.create(make_entry()) + assert created.view_count == 0 + + await repo.increment_view_count(created.id) + await repo.increment_view_count(created.id) + await db_session.flush() + + fetched = await repo.get_by_id(created.id) + assert fetched is not None + assert fetched.view_count == 2 + + @pytest.mark.asyncio + async def test_increment_view_count_nonexistent_returns_false(self, db_session): + repo = KnowledgeDBRepository(db_session) + + result = await repo.increment_view_count("00000000-0000-0000-0000-000000000000") + + assert result is False