logan
c7fccf99c7
P1 查询历史持久化 (47 tests): - QueryHistory/QuickQuery SQLAlchemy model 建表 - query_history_service 从内存迁移到 SQLite async - 每用户500条上限自动清理 - 7个API endpoint全部对接持久化 - 10个预置快捷查询模板 P2 EXPLAIN真实解析 (40 tests): - MySQL EXPLAIN FORMAT=JSON 执行+解析 - PostgreSQL EXPLAIN (FORMAT JSON) 执行+解析 - SQLite EXPLAIN QUERY PLAN 解析 - 统一树结构输出 + 优化警告 - POST /api/v1/sql-assistant/explain endpoint Dashboard修复: - 修复重复export default导致页面崩溃 - 对接/instances /tickets /dashboard/stats真实API - 替换硬编码统计数字为API驱动 - 修复calcHealthScore安全访问 AuditLogsPage: - 修复Fragment key warning 1081 tests passed (10 system_settings failures are pre-existing)
769 lines
29 KiB
Python
769 lines
29 KiB
Python
"""查询历史和快捷查询 持久化测试 — 验证 SQLite 存储。
|
|
|
|
测试范围:
|
|
- 查询历史记录和查看（CRUD + 分页 + 过滤）
|
|
- 收藏切换
|
|
- 快捷查询 CRUD
|
|
- 查询统计
|
|
- 权限控制
|
|
- 预置模板
|
|
"""
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.security import create_access_token, hash_password
|
|
from app.models.user import User
|
|
from app.models.query_history import QueryHistory, QuickQueryModel
|
|
from app.services.query_history_service import QueryHistoryService
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def admin_headers(admin_token: str) -> dict:
|
|
return {"Authorization": f"Bearer {admin_token}"}
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def developer_headers(developer_token: str) -> dict:
|
|
return {"Authorization": f"Bearer {developer_token}"}
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def guest_headers(guest_token: str) -> dict:
|
|
return {"Authorization": f"Bearer {guest_token}"}
|
|
|
|
|
|
# ── 查询历史记录 ──
|
|
|
|
|
|
class TestQueryHistoryRecord:
|
|
"""查询历史记录功能测试。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_query(self, db: AsyncSession):
|
|
"""测试记录一条查询。"""
|
|
entry = await QueryHistoryService.record_query(
|
|
db=db,
|
|
user_id=1,
|
|
username="admin",
|
|
instance_id=1,
|
|
instance_name="测试MySQL",
|
|
db_type="mysql",
|
|
database_name="test_db",
|
|
query_content="SELECT * FROM users LIMIT 10",
|
|
query_type="SELECT",
|
|
execution_time_ms=50,
|
|
result_rows=10,
|
|
)
|
|
await db.commit()
|
|
assert entry.id > 0
|
|
assert entry.user_id == 1
|
|
assert entry.db_type == "mysql"
|
|
assert entry.query_type == "SELECT"
|
|
assert entry.execution_time_ms == 50
|
|
assert entry.result_rows == 10
|
|
assert entry.is_success is True
|
|
assert entry.error_message is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_failed_query(self, db: AsyncSession):
|
|
"""测试记录失败的查询。"""
|
|
entry = await QueryHistoryService.record_query(
|
|
db=db,
|
|
user_id=1,
|
|
username="admin",
|
|
instance_id=1,
|
|
instance_name="测试MySQL",
|
|
db_type="mysql",
|
|
database_name="test_db",
|
|
query_content="SELECT * FROM nonexistent",
|
|
query_type="SELECT",
|
|
execution_time_ms=10,
|
|
result_rows=0,
|
|
is_success=False,
|
|
error_message="Table doesn't exist",
|
|
)
|
|
await db.commit()
|
|
assert entry.is_success is False
|
|
assert entry.error_message == "Table doesn't exist"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_multiple_db_types(self, db: AsyncSession):
|
|
"""测试记录不同数据库类型的查询。"""
|
|
for db_type, query, q_type in [
|
|
("mysql", "SELECT 1", "SELECT"),
|
|
("postgres", "SELECT version()", "SELECT"),
|
|
("redis", "GET mykey", "GET"),
|
|
("mongo", '{"collection": "users", "operation": "find"}', "find"),
|
|
]:
|
|
entry = await QueryHistoryService.record_query(
|
|
db=db,
|
|
user_id=1,
|
|
username="admin",
|
|
instance_id=1,
|
|
instance_name="测试",
|
|
db_type=db_type,
|
|
database_name="test",
|
|
query_content=query,
|
|
query_type=q_type,
|
|
execution_time_ms=10,
|
|
result_rows=1,
|
|
)
|
|
await db.commit()
|
|
assert entry.db_type == db_type
|
|
assert entry.query_type == q_type
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_query_long_content_truncated(self, db: AsyncSession):
|
|
"""测试超长查询内容被截断。"""
|
|
long_sql = "SELECT * FROM t WHERE id = " + "1" * 3000
|
|
entry = await QueryHistoryService.record_query(
|
|
db=db,
|
|
user_id=1,
|
|
username="admin",
|
|
instance_id=1,
|
|
instance_name="测试",
|
|
db_type="mysql",
|
|
database_name="test",
|
|
query_content=long_sql,
|
|
query_type="SELECT",
|
|
execution_time_ms=10,
|
|
result_rows=0,
|
|
)
|
|
await db.commit()
|
|
assert len(entry.query_content) == 2000
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_query_max_history_per_user(self, db: AsyncSession):
|
|
"""测试每个用户最多保留500条记录。"""
|
|
# 创建超过限制的记录
|
|
for i in range(510):
|
|
await QueryHistoryService.record_query(
|
|
db=db,
|
|
user_id=99,
|
|
username="test_user",
|
|
instance_id=1,
|
|
instance_name="测试",
|
|
db_type="mysql",
|
|
database_name="test",
|
|
query_content=f"SELECT {i}",
|
|
query_type="SELECT",
|
|
execution_time_ms=i,
|
|
result_rows=0,
|
|
)
|
|
await db.commit()
|
|
|
|
from sqlalchemy import select, func
|
|
count_stmt = select(func.count()).select_from(QueryHistory).where(
|
|
QueryHistory.user_id == 99
|
|
)
|
|
total = (await db.execute(count_stmt)).scalar()
|
|
assert total <= 500
|
|
|
|
|
|
# ── 查询历史查看 ──
|
|
|
|
|
|
class TestQueryHistoryView:
|
|
"""查询历史查看功能测试。"""
|
|
|
|
async def _setup_history(self, db: AsyncSession):
|
|
"""准备测试数据。"""
|
|
entries = [
|
|
("SELECT * FROM users", "mysql", 1, 50, 10),
|
|
("SELECT * FROM orders", "mysql", 1, 30, 20),
|
|
("GET mykey", "redis", 2, 5, 1),
|
|
("SELECT version()", "postgres", 3, 15, 1),
|
|
]
|
|
for sql, db_type, inst_id, time_ms, rows in entries:
|
|
await QueryHistoryService.record_query(
|
|
db=db,
|
|
user_id=1,
|
|
username="admin",
|
|
instance_id=inst_id,
|
|
instance_name=f"测试{db_type}",
|
|
db_type=db_type,
|
|
database_name="test",
|
|
query_content=sql,
|
|
query_type="SELECT" if db_type != "redis" else "GET",
|
|
execution_time_ms=time_ms,
|
|
result_rows=rows,
|
|
)
|
|
await db.commit()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_history_default(self, db: AsyncSession):
|
|
"""测试默认查看历史。"""
|
|
await self._setup_history(db)
|
|
result = await QueryHistoryService.get_history(db=db, user_id=1)
|
|
assert result["total"] == 4
|
|
assert result["page"] == 1
|
|
assert result["page_size"] == 20
|
|
assert len(result["items"]) == 4
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_history_filter_by_db_type(self, db: AsyncSession):
|
|
"""测试按数据库类型筛选。"""
|
|
await self._setup_history(db)
|
|
result = await QueryHistoryService.get_history(db=db, user_id=1, db_type="mysql")
|
|
assert result["total"] == 2
|
|
for item in result["items"]:
|
|
assert item["db_type"] == "mysql"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_history_filter_by_instance(self, db: AsyncSession):
|
|
"""测试按实例ID筛选。"""
|
|
await self._setup_history(db)
|
|
result = await QueryHistoryService.get_history(db=db, user_id=1, instance_id=2)
|
|
assert result["total"] == 1
|
|
assert result["items"][0]["db_type"] == "redis"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_history_filter_by_keyword(self, db: AsyncSession):
|
|
"""测试按关键词筛选。"""
|
|
await self._setup_history(db)
|
|
result = await QueryHistoryService.get_history(db=db, user_id=1, keyword="users")
|
|
assert result["total"] == 1
|
|
assert "users" in result["items"][0]["query_content"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_history_pagination(self, db: AsyncSession):
|
|
"""测试分页。"""
|
|
await self._setup_history(db)
|
|
result = await QueryHistoryService.get_history(db=db, user_id=1, page=1, page_size=2)
|
|
assert result["total"] == 4
|
|
assert len(result["items"]) == 2
|
|
|
|
result2 = await QueryHistoryService.get_history(db=db, user_id=1, page=2, page_size=2)
|
|
assert result2["total"] == 4
|
|
assert len(result2["items"]) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_history_other_user_empty(self, db: AsyncSession):
|
|
"""测试查看其他用户的历史为空。"""
|
|
await self._setup_history(db)
|
|
result = await QueryHistoryService.get_history(db=db, user_id=999)
|
|
assert result["total"] == 0
|
|
assert result["items"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_history_sorted_by_time_desc(self, db: AsyncSession):
|
|
"""测试历史按时间倒序排列。"""
|
|
await self._setup_history(db)
|
|
result = await QueryHistoryService.get_history(db=db, user_id=1)
|
|
times = [item["created_at"] for item in result["items"]]
|
|
assert times == sorted(times, reverse=True)
|
|
|
|
|
|
# ── 收藏功能 ──
|
|
|
|
|
|
class TestQueryHistoryFavorite:
|
|
"""查询收藏功能测试。"""
|
|
|
|
async def _setup_for_favorite(self, db: AsyncSession):
|
|
entry = await QueryHistoryService.record_query(
|
|
db=db,
|
|
user_id=1, username="admin", instance_id=1,
|
|
instance_name="测试", db_type="mysql", database_name="test",
|
|
query_content="SELECT 1", query_type="SELECT",
|
|
execution_time_ms=10, result_rows=1,
|
|
)
|
|
await db.commit()
|
|
return entry.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_favorite_add(self, db: AsyncSession):
|
|
"""测试添加收藏。"""
|
|
entry_id = await self._setup_for_favorite(db)
|
|
is_fav = await QueryHistoryService.toggle_favorite(db=db, user_id=1, history_id=entry_id)
|
|
await db.commit()
|
|
assert is_fav is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_favorite_remove(self, db: AsyncSession):
|
|
"""测试取消收藏。"""
|
|
entry_id = await self._setup_for_favorite(db)
|
|
await QueryHistoryService.toggle_favorite(db=db, user_id=1, history_id=entry_id)
|
|
await db.commit()
|
|
is_fav = await QueryHistoryService.toggle_favorite(db=db, user_id=1, history_id=entry_id)
|
|
await db.commit()
|
|
assert is_fav is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_favorites(self, db: AsyncSession):
|
|
"""测试获取收藏列表。"""
|
|
entry_id = await self._setup_for_favorite(db)
|
|
await QueryHistoryService.toggle_favorite(db=db, user_id=1, history_id=entry_id)
|
|
await db.commit()
|
|
|
|
result = await QueryHistoryService.get_favorites(db=db, user_id=1)
|
|
assert result["total"] == 1
|
|
assert result["items"][0]["id"] == entry_id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_favorites_empty(self, db: AsyncSession):
|
|
"""测试无收藏时返回空列表。"""
|
|
result = await QueryHistoryService.get_favorites(db=db, user_id=1)
|
|
assert result["total"] == 0
|
|
assert result["items"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_favorites_per_user(self, db: AsyncSession):
|
|
"""测试收藏是用户隔离的。"""
|
|
entry_id = await self._setup_for_favorite(db)
|
|
await QueryHistoryService.toggle_favorite(db=db, user_id=1, history_id=entry_id)
|
|
await db.commit()
|
|
|
|
result = await QueryHistoryService.get_favorites(db=db, user_id=999)
|
|
assert result["total"] == 0
|
|
|
|
|
|
# ── 快捷查询 ──
|
|
|
|
|
|
class TestQuickQueries:
|
|
"""快捷查询功能测试。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_quick_query(self, db: AsyncSession):
|
|
"""测试创建快捷查询。"""
|
|
qq = await QueryHistoryService.create_quick_query(
|
|
db=db,
|
|
name="查看活跃用户",
|
|
description="查看最近活跃的用户列表",
|
|
query_content="SELECT * FROM users WHERE last_login > DATE_SUB(NOW(), INTERVAL 7 DAY)",
|
|
db_type="mysql",
|
|
database_pattern="*",
|
|
is_shared=False,
|
|
created_by=1,
|
|
created_by_name="admin",
|
|
)
|
|
await db.commit()
|
|
assert qq.id > 0
|
|
assert qq.name == "查看活跃用户"
|
|
assert qq.is_shared is False
|
|
assert qq.created_by == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_quick_queries_personal(self, db: AsyncSession):
|
|
"""测试获取个人快捷查询。"""
|
|
await QueryHistoryService.create_quick_query(
|
|
db=db, name="个人查询", description="", query_content="SELECT 1",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=False, created_by=1, created_by_name="admin",
|
|
)
|
|
await QueryHistoryService.create_quick_query(
|
|
db=db, name="他人查询", description="", query_content="SELECT 2",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=False, created_by=2, created_by_name="dba",
|
|
)
|
|
await db.commit()
|
|
|
|
# 用户1只能看到自己的 + 共享的
|
|
items = await QueryHistoryService.get_quick_queries(db=db, user_id=1)
|
|
assert len(items) == 1
|
|
assert items[0]["name"] == "个人查询"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_quick_queries_shared(self, db: AsyncSession):
|
|
"""测试获取共享快捷查询。"""
|
|
await QueryHistoryService.create_quick_query(
|
|
db=db, name="共享查询", description="", query_content="SELECT 1",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=True, created_by=2, created_by_name="dba",
|
|
)
|
|
await db.commit()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(db=db, user_id=1)
|
|
assert len(items) == 1
|
|
assert items[0]["is_shared"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_quick_queries_filter_db_type(self, db: AsyncSession):
|
|
"""测试按数据库类型筛选快捷查询。"""
|
|
await QueryHistoryService.create_quick_query(
|
|
db=db, name="MySQL查询", description="", query_content="SELECT 1",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=True, created_by=1, created_by_name="admin",
|
|
)
|
|
await QueryHistoryService.create_quick_query(
|
|
db=db, name="Redis查询", description="", query_content="GET key",
|
|
db_type="redis", database_pattern="*",
|
|
is_shared=True, created_by=1, created_by_name="admin",
|
|
)
|
|
await db.commit()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(db=db, user_id=1, db_type="redis")
|
|
assert len(items) == 1
|
|
assert items[0]["db_type"] == "redis"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_quick_queries_filter_keyword(self, db: AsyncSession):
|
|
"""测试按关键词筛选快捷查询。"""
|
|
await QueryHistoryService.create_quick_query(
|
|
db=db, name="查看活跃用户", description="", query_content="SELECT 1",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=True, created_by=1, created_by_name="admin",
|
|
)
|
|
await QueryHistoryService.create_quick_query(
|
|
db=db, name="查看慢查询", description="", query_content="SELECT 2",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=True, created_by=1, created_by_name="admin",
|
|
)
|
|
await db.commit()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(db=db, user_id=1, keyword="活跃")
|
|
assert len(items) == 1
|
|
assert "活跃" in items[0]["name"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_use_quick_query(self, db: AsyncSession):
|
|
"""测试使用快捷查询。"""
|
|
qq = await QueryHistoryService.create_quick_query(
|
|
db=db, name="测试", description="", query_content="SELECT 1",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=True, created_by=1, created_by_name="admin",
|
|
)
|
|
await db.commit()
|
|
|
|
content = await QueryHistoryService.use_quick_query(db, qq.id)
|
|
await db.commit()
|
|
assert content == "SELECT 1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_use_quick_query_increments_count(self, db: AsyncSession):
|
|
"""测试使用快捷查询增加使用次数。"""
|
|
qq = await QueryHistoryService.create_quick_query(
|
|
db=db, name="测试", description="", query_content="SELECT 1",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=True, created_by=1, created_by_name="admin",
|
|
)
|
|
await db.commit()
|
|
assert qq.usage_count == 0
|
|
|
|
await QueryHistoryService.use_quick_query(db, qq.id)
|
|
await db.commit()
|
|
assert qq.usage_count == 1
|
|
|
|
await QueryHistoryService.use_quick_query(db, qq.id)
|
|
await db.commit()
|
|
assert qq.usage_count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_use_quick_query_not_found(self, db: AsyncSession):
|
|
"""测试使用不存在的快捷查询。"""
|
|
content = await QueryHistoryService.use_quick_query(db, 99999)
|
|
assert content is None
|
|
|
|
|
|
# ── 预置模板 ──
|
|
|
|
|
|
class TestDefaultQuickQueries:
|
|
"""预置快捷查询模板测试。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_templates_exist(self, db: AsyncSession):
|
|
"""测试预置模板存在。"""
|
|
await QueryHistoryService.seed_default_quick_queries(db)
|
|
await db.commit()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(db=db, user_id=1)
|
|
# 预置了10个模板
|
|
assert len(items) == 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_templates_cover_db_types(self, db: AsyncSession):
|
|
"""测试预置模板覆盖所有数据库类型。"""
|
|
await QueryHistoryService.seed_default_quick_queries(db)
|
|
await db.commit()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(db=db, user_id=1)
|
|
db_types = {item["db_type"] for item in items}
|
|
assert "mysql" in db_types
|
|
assert "redis" in db_types
|
|
assert "postgres" in db_types
|
|
assert "mongo" in db_types
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_templates_all_shared(self, db: AsyncSession):
|
|
"""测试预置模板都是共享的。"""
|
|
await QueryHistoryService.seed_default_quick_queries(db)
|
|
await db.commit()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(db=db, user_id=1)
|
|
for item in items:
|
|
assert item["is_shared"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_templates_system_created(self, db: AsyncSession):
|
|
"""测试预置模板由系统创建。"""
|
|
await QueryHistoryService.seed_default_quick_queries(db)
|
|
await db.commit()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(db=db, user_id=1)
|
|
for item in items:
|
|
assert item["created_by_name"] == "系统"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seed_idempotent(self, db: AsyncSession):
|
|
"""测试 seed 是幂等的（多次调用不会重复插入）。"""
|
|
await QueryHistoryService.seed_default_quick_queries(db)
|
|
await db.commit()
|
|
await QueryHistoryService.seed_default_quick_queries(db)
|
|
await db.commit()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(db=db, user_id=1)
|
|
assert len(items) == 10
|
|
|
|
|
|
# ── 查询统计 ──
|
|
|
|
|
|
class TestQueryStatistics:
|
|
"""查询统计功能测试。"""
|
|
|
|
async def _setup_stats(self, db: AsyncSession):
|
|
for i in range(5):
|
|
await QueryHistoryService.record_query(
|
|
db=db, user_id=1, username="admin", instance_id=1,
|
|
instance_name="测试", db_type="mysql", database_name="test",
|
|
query_content=f"SELECT * FROM table_{i} WHERE id = 1",
|
|
query_type="SELECT",
|
|
execution_time_ms=10 + i * 10,
|
|
result_rows=i + 1,
|
|
)
|
|
for i in range(3):
|
|
await QueryHistoryService.record_query(
|
|
db=db, user_id=1, username="admin", instance_id=2,
|
|
instance_name="测试Redis", db_type="redis", database_name="db0",
|
|
query_content="GET mykey",
|
|
query_type="GET",
|
|
execution_time_ms=5,
|
|
result_rows=1,
|
|
)
|
|
await db.commit()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_statistics_basic(self, db: AsyncSession):
|
|
"""测试基本统计。"""
|
|
await self._setup_stats(db)
|
|
stats = await QueryHistoryService.get_query_statistics(db=db, user_id=1)
|
|
|
|
assert stats["total_queries"] == 8
|
|
assert stats["avg_execution_time_ms"] > 0
|
|
assert stats["success_rate"] == 100.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_statistics_db_type_distribution(self, db: AsyncSession):
|
|
"""测试数据库类型分布统计。"""
|
|
await self._setup_stats(db)
|
|
stats = await QueryHistoryService.get_query_statistics(db=db, user_id=1)
|
|
|
|
assert "mysql" in stats["db_type_distribution"]
|
|
assert "redis" in stats["db_type_distribution"]
|
|
assert stats["db_type_distribution"]["mysql"] == 5
|
|
assert stats["db_type_distribution"]["redis"] == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_statistics_daily_trend(self, db: AsyncSession):
|
|
"""测试每日趋势。"""
|
|
await self._setup_stats(db)
|
|
stats = await QueryHistoryService.get_query_statistics(db=db, user_id=1)
|
|
|
|
assert "daily_trend" in stats
|
|
assert len(stats["daily_trend"]) == 7
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_statistics_empty(self, db: AsyncSession):
|
|
"""测试无记录时返回提示。"""
|
|
stats = await QueryHistoryService.get_query_statistics(db=db, user_id=999)
|
|
assert "message" in stats
|
|
assert stats["message"] == "暂无查询记录"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_statistics_top_tables(self, db: AsyncSession):
|
|
"""测试高频表统计。"""
|
|
await self._setup_stats(db)
|
|
stats = await QueryHistoryService.get_query_statistics(db=db, user_id=1)
|
|
|
|
assert "top_tables" in stats
|
|
# 应该能从SQL中提取表名
|
|
assert len(stats["top_tables"]) > 0
|
|
|
|
|
|
# ── API端点集成测试 ──
|
|
|
|
|
|
class TestQueryHistoryAPI:
|
|
"""查询历史 API 端点集成测试。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_my_history_unauthorized(self, client):
|
|
"""测试未认证访问历史。"""
|
|
resp = await client.get("/api/v1/query-history/my")
|
|
assert resp.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_my_history_empty(self, client, admin_headers, db):
|
|
"""测试空历史记录。"""
|
|
resp = await client.get("/api/v1/query-history/my", headers=admin_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 0
|
|
assert data["items"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_my_history_with_data(self, client, admin_headers, db, admin_user):
|
|
"""测试有数据的历史记录。"""
|
|
await QueryHistoryService.record_query(
|
|
db=db, user_id=admin_user.id, username="admin", instance_id=1,
|
|
instance_name="测试", db_type="mysql", database_name="test",
|
|
query_content="SELECT 1", query_type="SELECT",
|
|
execution_time_ms=10, result_rows=1,
|
|
)
|
|
await db.commit()
|
|
|
|
resp = await client.get("/api/v1/query-history/my", headers=admin_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] >= 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_my_history_filter_db_type(self, client, admin_headers, db, admin_user):
|
|
"""测试按类型筛选。"""
|
|
await QueryHistoryService.record_query(
|
|
db=db, user_id=admin_user.id, username="admin", instance_id=1,
|
|
instance_name="测试", db_type="mysql", database_name="test",
|
|
query_content="SELECT 1", query_type="SELECT",
|
|
execution_time_ms=10, result_rows=1,
|
|
)
|
|
await QueryHistoryService.record_query(
|
|
db=db, user_id=admin_user.id, username="admin", instance_id=2,
|
|
instance_name="测试Redis", db_type="redis", database_name="db0",
|
|
query_content="GET key", query_type="GET",
|
|
execution_time_ms=5, result_rows=1,
|
|
)
|
|
await db.commit()
|
|
|
|
resp = await client.get(
|
|
"/api/v1/query-history/my?db_type=mysql",
|
|
headers=admin_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 1
|
|
assert data["items"][0]["db_type"] == "mysql"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_favorite_api(self, client, admin_headers, db, admin_user):
|
|
"""测试收藏API。"""
|
|
entry = await QueryHistoryService.record_query(
|
|
db=db, user_id=admin_user.id, username="admin", instance_id=1,
|
|
instance_name="测试", db_type="mysql", database_name="test",
|
|
query_content="SELECT 1", query_type="SELECT",
|
|
execution_time_ms=10, result_rows=1,
|
|
)
|
|
await db.commit()
|
|
|
|
resp = await client.post(
|
|
f"/api/v1/query-history/{entry.id}/favorite",
|
|
headers=admin_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["is_favorite"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_favorites_api(self, client, admin_headers, db):
|
|
"""测试收藏列表API。"""
|
|
resp = await client.get("/api/v1/query-history/favorites", headers=admin_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "total" in data
|
|
assert "items" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_statistics_api(self, client, admin_headers, db):
|
|
"""测试统计API。"""
|
|
resp = await client.get("/api/v1/query-history/statistics", headers=admin_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "message" in data or "total_queries" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_quick_queries_list_api(self, client, admin_headers, db):
|
|
"""测试快捷查询列表API。"""
|
|
await QueryHistoryService.seed_default_quick_queries(db)
|
|
await db.commit()
|
|
|
|
resp = await client.get("/api/v1/query-history/quick-queries", headers=admin_headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 10
|
|
assert len(data["items"]) == 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_quick_queries_create_api(self, client, admin_headers, db):
|
|
"""测试创建快捷查询API。"""
|
|
resp = await client.post(
|
|
"/api/v1/query-history/quick-queries",
|
|
params={
|
|
"name": "我的查询",
|
|
"query_content": "SELECT * FROM users",
|
|
"db_type": "mysql",
|
|
"description": "测试快捷查询",
|
|
"is_shared": False,
|
|
},
|
|
headers=admin_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["name"] == "我的查询"
|
|
assert "id" in data
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_quick_queries_filter_api(self, client, admin_headers, db):
|
|
"""测试快捷查询筛选API。"""
|
|
await QueryHistoryService.seed_default_quick_queries(db)
|
|
await db.commit()
|
|
|
|
resp = await client.get(
|
|
"/api/v1/query-history/quick-queries?db_type=redis",
|
|
headers=admin_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
for item in data["items"]:
|
|
assert item["db_type"] == "redis"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_guest_can_view_own_history(self, client, guest_headers, db):
|
|
"""测试访客可以查看自己的历史（但无权执行查询）。"""
|
|
resp = await client.get("/api/v1/query-history/my", headers=guest_headers)
|
|
assert resp.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pagination_parameters(self, client, admin_headers, db, admin_user):
|
|
"""测试分页参数。"""
|
|
for i in range(15):
|
|
await QueryHistoryService.record_query(
|
|
db=db, user_id=admin_user.id, username="admin", instance_id=1,
|
|
instance_name="测试", db_type="mysql", database_name="test",
|
|
query_content=f"SELECT {i}", query_type="SELECT",
|
|
execution_time_ms=10, result_rows=1,
|
|
)
|
|
await db.commit()
|
|
|
|
resp = await client.get(
|
|
"/api/v1/query-history/my?page=1&page_size=5",
|
|
headers=admin_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 15
|
|
assert len(data["items"]) == 5
|
|
assert data["page"] == 1
|
|
assert data["page_size"] == 5
|