Gan, Jimmy
b981c06d59
- Fix unit test imports: add env setup in conftest.py before module imports - Add 24 new auth router tests (RBAC, preferences, password validation) - Add 16 new tests for litellm and chat_summary routers - Apply black formatting and ruff linting across codebase - Add pre-commit hooks configuration (black, ruff, file checks) - Increase CI coverage threshold from 40% to 50% Test Results: - 206 tests passing (91 unit + 115 integration) - Coverage: 58.79% on core modules - auth.py: 57% → 85%, litellm.py: 23% → 87%, chat_summary.py: 41% → 100% - auth_service: 96.51%, config: 100%, rbac: 93.48%
201 lines
7.9 KiB
Python
201 lines
7.9 KiB
Python
"""
|
|
Integration tests for chat summary operations.
|
|
"""
|
|
import pytest
|
|
import aiosqlite
|
|
import os
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
|
|
|
|
class TestChatSummaryDates:
|
|
"""Test chat summary dates endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_dates_empty(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test listing dates when database is empty."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create empty database with schema
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE summaries (date TEXT, content TEXT, created_at TEXT)")
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/dates", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_dates_with_data(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test listing dates with summaries."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create database with test data
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE summaries (date TEXT, content TEXT, created_at TEXT)")
|
|
await db.execute(
|
|
"INSERT INTO summaries VALUES (?, ?, ?)", ("2024-01-01", "Summary 1", "2024-01-01T10:00:00")
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO summaries VALUES (?, ?, ?)", ("2024-01-02", "Summary 2", "2024-01-02T10:00:00")
|
|
)
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/dates", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 2
|
|
assert "2024-01-02" in data
|
|
assert "2024-01-01" in data
|
|
|
|
def test_list_dates_without_auth(self, test_app, mock_config):
|
|
"""Test listing dates without authentication."""
|
|
response = test_app.get("/api/chat-summary/dates")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestChatSummaryGet:
|
|
"""Test get summary endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_summary_success(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test getting a summary for a specific date."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create database with test data
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE summaries (date TEXT, content TEXT, created_at TEXT)")
|
|
await db.execute(
|
|
"INSERT INTO summaries VALUES (?, ?, ?)",
|
|
("2024-01-01", "Test summary content", "2024-01-01T10:00:00"),
|
|
)
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/summary/2024-01-01", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["date"] == "2024-01-01"
|
|
assert data["content"] == "Test summary content"
|
|
assert data["created_at"] == "2024-01-01T10:00:00"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_summary_not_found(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test getting a summary for a date that doesn't exist."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create empty database
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE summaries (date TEXT, content TEXT, created_at TEXT)")
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/summary/2024-01-01", headers=admin_headers)
|
|
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "No summary for this date" in data["detail"]
|
|
|
|
def test_get_summary_without_auth(self, test_app, mock_config):
|
|
"""Test getting summary without authentication."""
|
|
response = test_app.get("/api/chat-summary/summary/2024-01-01")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestChatSummaryMessages:
|
|
"""Test get messages endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_messages_success(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test getting messages for a specific date."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create database with test data
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE messages (group_name TEXT, sender_name TEXT, text TEXT, timestamp TEXT)")
|
|
await db.execute(
|
|
"INSERT INTO messages VALUES (?, ?, ?, ?)",
|
|
("Group1", "User1", "Hello", "2024-01-01 10:00:00"),
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO messages VALUES (?, ?, ?, ?)",
|
|
("Group1", "User2", "Hi there", "2024-01-01 10:01:00"),
|
|
)
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/messages/2024-01-01", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 2
|
|
assert data[0]["group"] == "Group1"
|
|
assert data[0]["sender"] == "User1"
|
|
assert data[0]["text"] == "Hello"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_messages_empty(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test getting messages when none exist for date."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create empty database
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE messages (group_name TEXT, sender_name TEXT, text TEXT, timestamp TEXT)")
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/messages/2024-01-01", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) == 0
|
|
|
|
def test_get_messages_without_auth(self, test_app, mock_config):
|
|
"""Test getting messages without authentication."""
|
|
response = test_app.get("/api/chat-summary/messages/2024-01-01")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestChatSummaryTrigger:
|
|
"""Test trigger summary endpoint."""
|
|
|
|
def test_trigger_summary_success(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test triggering a summary generation."""
|
|
trigger_path = tmp_path / "trigger"
|
|
|
|
with patch.dict(os.environ, {"CHAT_SUMMARY_TRIGGER": str(trigger_path)}):
|
|
response = test_app.post("/api/chat-summary/trigger", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "triggered"
|
|
assert trigger_path.exists()
|
|
assert trigger_path.read_text() == "1"
|
|
|
|
def test_trigger_summary_creates_directory(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test that trigger creates parent directory if needed."""
|
|
trigger_path = tmp_path / "subdir" / "trigger"
|
|
|
|
with patch.dict(os.environ, {"CHAT_SUMMARY_TRIGGER": str(trigger_path)}):
|
|
response = test_app.post("/api/chat-summary/trigger", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
assert trigger_path.parent.exists()
|
|
assert trigger_path.exists()
|
|
|
|
def test_trigger_summary_without_auth(self, test_app, mock_config):
|
|
"""Test triggering summary without authentication."""
|
|
response = test_app.post("/api/chat-summary/trigger")
|
|
|
|
assert response.status_code == 401
|