5e836bde24
All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 7m18s
- tests/integration/conftest.py: 連接 awoooi_dev PostgreSQL，每個測試後 rollback - tests/integration/test_knowledge_repository.py: 23 個真實 DB 測試 - create/get_by_id/list/update/delete(軟刪除)/search/categories/view_count - tests/integration/test_incident_api.py: 7 個 HTTPS 端點測試 - health check + knowledge API smoke test - 遵循禁止 Mock 鐵律 (feedback_no_mock_testing.md) - 本地驗證: 30/30 PASSED Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
305 lines
10 KiB
Python
305 lines
10 KiB
Python
"""
|
||
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
|