logan
4c9427e546
- P0: 实例访问控制全线接入权限组 (829测试) - P1: permissions字段激活 + 操作级权限检查 (843测试) - 前端: 删除数据权限页面, 权限组合并为唯一入口 - 新增: CodeMirror编辑器 + 查询模板 + 慢查询分析 - 新增: MongoDB管理页面 + Dashboard增强 - 修复: Pydantic V2迁移 + 前端TS错误修复
794 lines
29 KiB
Python
794 lines
29 KiB
Python
"""查询历史和快捷查询 API 测试 — P1 审计闭环。
|
|
|
|
测试范围:
|
|
- 查询历史记录和查看
|
|
- 收藏切换
|
|
- 快捷查询 CRUD
|
|
- 查询统计
|
|
- 权限控制
|
|
- 预置模板
|
|
"""
|
|
import pytest
|
|
import pytest_asyncio
|
|
from httpx import AsyncClient
|
|
|
|
from app.core.security import create_access_token, hash_password
|
|
from app.models.user import User
|
|
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):
|
|
"""测试记录一条查询。"""
|
|
entry = await QueryHistoryService.record_query(
|
|
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,
|
|
)
|
|
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):
|
|
"""测试记录失败的查询。"""
|
|
entry = await QueryHistoryService.record_query(
|
|
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",
|
|
)
|
|
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):
|
|
"""测试记录不同数据库类型的查询。"""
|
|
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(
|
|
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,
|
|
)
|
|
assert entry.db_type == db_type
|
|
assert entry.query_type == q_type
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_query_long_content_truncated(self):
|
|
"""测试超长查询内容被截断。"""
|
|
long_sql = "SELECT * FROM t WHERE id = " + "1" * 3000
|
|
entry = await QueryHistoryService.record_query(
|
|
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,
|
|
)
|
|
assert len(entry.query_content) == 2000
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_query_max_history_per_user(self):
|
|
"""测试每个用户最多保留500条记录。"""
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
# 创建超过限制的记录
|
|
for i in range(510):
|
|
await QueryHistoryService.record_query(
|
|
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,
|
|
)
|
|
|
|
user_entries = [h for h in QueryHistoryService._history if h.user_id == 99]
|
|
assert len(user_entries) <= 500
|
|
|
|
# 清理
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
|
|
|
|
# ── 查询历史查看 ──
|
|
|
|
|
|
class TestQueryHistoryView:
|
|
"""查询历史查看功能测试。"""
|
|
|
|
async def _setup_history(self):
|
|
"""准备测试数据。"""
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
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(
|
|
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,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_history_default(self):
|
|
"""测试默认查看历史。"""
|
|
await self._setup_history()
|
|
result = await QueryHistoryService.get_history(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):
|
|
"""测试按数据库类型筛选。"""
|
|
await self._setup_history()
|
|
result = await QueryHistoryService.get_history(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):
|
|
"""测试按实例ID筛选。"""
|
|
await self._setup_history()
|
|
result = await QueryHistoryService.get_history(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):
|
|
"""测试按关键词筛选。"""
|
|
await self._setup_history()
|
|
result = await QueryHistoryService.get_history(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):
|
|
"""测试分页。"""
|
|
await self._setup_history()
|
|
result = await QueryHistoryService.get_history(user_id=1, page=1, page_size=2)
|
|
assert result["total"] == 4
|
|
assert len(result["items"]) == 2
|
|
|
|
result2 = await QueryHistoryService.get_history(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):
|
|
"""测试查看其他用户的历史为空。"""
|
|
await self._setup_history()
|
|
result = await QueryHistoryService.get_history(user_id=999)
|
|
assert result["total"] == 0
|
|
assert result["items"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_history_sorted_by_time_desc(self):
|
|
"""测试历史按时间倒序排列。"""
|
|
await self._setup_history()
|
|
result = await QueryHistoryService.get_history(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):
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
entry = await QueryHistoryService.record_query(
|
|
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,
|
|
)
|
|
return entry.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_favorite_add(self):
|
|
"""测试添加收藏。"""
|
|
entry_id = await self._setup_for_favorite()
|
|
is_fav = await QueryHistoryService.toggle_favorite(user_id=1, history_id=entry_id)
|
|
assert is_fav is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_toggle_favorite_remove(self):
|
|
"""测试取消收藏。"""
|
|
entry_id = await self._setup_for_favorite()
|
|
await QueryHistoryService.toggle_favorite(user_id=1, history_id=entry_id)
|
|
is_fav = await QueryHistoryService.toggle_favorite(user_id=1, history_id=entry_id)
|
|
assert is_fav is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_favorites(self):
|
|
"""测试获取收藏列表。"""
|
|
entry_id = await self._setup_for_favorite()
|
|
await QueryHistoryService.toggle_favorite(user_id=1, history_id=entry_id)
|
|
|
|
result = await QueryHistoryService.get_favorites(user_id=1)
|
|
assert result["total"] == 1
|
|
assert result["items"][0]["id"] == entry_id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_favorites_empty(self):
|
|
"""测试无收藏时返回空列表。"""
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._favorites = set()
|
|
|
|
result = await QueryHistoryService.get_favorites(user_id=1)
|
|
assert result["total"] == 0
|
|
assert result["items"] == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_favorites_per_user(self):
|
|
"""测试收藏是用户隔离的。"""
|
|
entry_id = await self._setup_for_favorite()
|
|
await QueryHistoryService.toggle_favorite(user_id=1, history_id=entry_id)
|
|
|
|
result = await QueryHistoryService.get_favorites(user_id=999)
|
|
assert result["total"] == 0
|
|
|
|
|
|
# ── 快捷查询 ──
|
|
|
|
|
|
class TestQuickQueries:
|
|
"""快捷查询功能测试。"""
|
|
|
|
async def _setup(self):
|
|
QueryHistoryService._quick_queries = []
|
|
QueryHistoryService._quick_counter = 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_quick_query(self):
|
|
"""测试创建快捷查询。"""
|
|
await self._setup()
|
|
qq = await QueryHistoryService.create_quick_query(
|
|
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",
|
|
)
|
|
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):
|
|
"""测试获取个人快捷查询。"""
|
|
await self._setup()
|
|
await QueryHistoryService.create_quick_query(
|
|
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(
|
|
name="他人查询", description="", query_content="SELECT 2",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=False, created_by=2, created_by_name="dba",
|
|
)
|
|
|
|
# 用户1只能看到自己的 + 共享的
|
|
items = await QueryHistoryService.get_quick_queries(user_id=1)
|
|
assert len(items) == 1
|
|
assert items[0]["name"] == "个人查询"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_quick_queries_shared(self):
|
|
"""测试获取共享快捷查询。"""
|
|
await self._setup()
|
|
await QueryHistoryService.create_quick_query(
|
|
name="共享查询", description="", query_content="SELECT 1",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=True, created_by=2, created_by_name="dba",
|
|
)
|
|
|
|
items = await QueryHistoryService.get_quick_queries(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):
|
|
"""测试按数据库类型筛选快捷查询。"""
|
|
await self._setup()
|
|
await QueryHistoryService.create_quick_query(
|
|
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(
|
|
name="Redis查询", description="", query_content="GET key",
|
|
db_type="redis", database_pattern="*",
|
|
is_shared=True, created_by=1, created_by_name="admin",
|
|
)
|
|
|
|
items = await QueryHistoryService.get_quick_queries(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):
|
|
"""测试按关键词筛选快捷查询。"""
|
|
await self._setup()
|
|
await QueryHistoryService.create_quick_query(
|
|
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(
|
|
name="查看慢查询", description="", query_content="SELECT 2",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=True, created_by=1, created_by_name="admin",
|
|
)
|
|
|
|
items = await QueryHistoryService.get_quick_queries(user_id=1, keyword="活跃")
|
|
assert len(items) == 1
|
|
assert "活跃" in items[0]["name"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_use_quick_query(self):
|
|
"""测试使用快捷查询。"""
|
|
await self._setup()
|
|
qq = await QueryHistoryService.create_quick_query(
|
|
name="测试", description="", query_content="SELECT 1",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=True, created_by=1, created_by_name="admin",
|
|
)
|
|
|
|
content = await QueryHistoryService.use_quick_query(qq.id)
|
|
assert content == "SELECT 1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_use_quick_query_increments_count(self):
|
|
"""测试使用快捷查询增加使用次数。"""
|
|
await self._setup()
|
|
qq = await QueryHistoryService.create_quick_query(
|
|
name="测试", description="", query_content="SELECT 1",
|
|
db_type="mysql", database_pattern="*",
|
|
is_shared=True, created_by=1, created_by_name="admin",
|
|
)
|
|
assert qq.usage_count == 0
|
|
|
|
await QueryHistoryService.use_quick_query(qq.id)
|
|
assert qq.usage_count == 1
|
|
|
|
await QueryHistoryService.use_quick_query(qq.id)
|
|
assert qq.usage_count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_use_quick_query_not_found(self):
|
|
"""测试使用不存在的快捷查询。"""
|
|
content = await QueryHistoryService.use_quick_query(99999)
|
|
assert content is None
|
|
|
|
|
|
# ── 预置模板 ──
|
|
|
|
|
|
class TestDefaultQuickQueries:
|
|
"""预置快捷查询模板测试。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_templates_exist(self):
|
|
"""测试预置模板存在。"""
|
|
QueryHistoryService._quick_queries = []
|
|
QueryHistoryService._quick_counter = 0
|
|
QueryHistoryService.seed_default_quick_queries()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(user_id=1)
|
|
# 预置了10个模板
|
|
assert len(items) == 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_templates_cover_db_types(self):
|
|
"""测试预置模板覆盖所有数据库类型。"""
|
|
QueryHistoryService._quick_queries = []
|
|
QueryHistoryService._quick_counter = 0
|
|
QueryHistoryService.seed_default_quick_queries()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(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):
|
|
"""测试预置模板都是共享的。"""
|
|
QueryHistoryService._quick_queries = []
|
|
QueryHistoryService._quick_counter = 0
|
|
QueryHistoryService.seed_default_quick_queries()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(user_id=1)
|
|
for item in items:
|
|
assert item["is_shared"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_default_templates_system_created(self):
|
|
"""测试预置模板由系统创建。"""
|
|
QueryHistoryService._quick_queries = []
|
|
QueryHistoryService._quick_counter = 0
|
|
QueryHistoryService.seed_default_quick_queries()
|
|
|
|
items = await QueryHistoryService.get_quick_queries(user_id=1)
|
|
for item in items:
|
|
assert item["created_by_name"] == "系统"
|
|
|
|
|
|
# ── 查询统计 ──
|
|
|
|
|
|
class TestQueryStatistics:
|
|
"""查询统计功能测试。"""
|
|
|
|
async def _setup_stats(self):
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
for i in range(5):
|
|
await QueryHistoryService.record_query(
|
|
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(
|
|
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,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_statistics_basic(self):
|
|
"""测试基本统计。"""
|
|
await self._setup_stats()
|
|
stats = await QueryHistoryService.get_query_statistics(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):
|
|
"""测试数据库类型分布统计。"""
|
|
await self._setup_stats()
|
|
stats = await QueryHistoryService.get_query_statistics(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):
|
|
"""测试每日趋势。"""
|
|
await self._setup_stats()
|
|
stats = await QueryHistoryService.get_query_statistics(user_id=1)
|
|
|
|
assert "daily_trend" in stats
|
|
assert len(stats["daily_trend"]) == 7
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_statistics_empty(self):
|
|
"""测试无记录时返回提示。"""
|
|
QueryHistoryService._history = []
|
|
stats = await QueryHistoryService.get_query_statistics(user_id=999)
|
|
assert "message" in stats
|
|
assert stats["message"] == "暂无查询记录"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_statistics_top_tables(self):
|
|
"""测试高频表统计。"""
|
|
await self._setup_stats()
|
|
stats = await QueryHistoryService.get_query_statistics(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: AsyncClient):
|
|
"""测试未认证访问历史。"""
|
|
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: AsyncClient, admin_headers: dict):
|
|
"""测试空历史记录。"""
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
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: AsyncClient, admin_headers: dict):
|
|
"""测试有数据的历史记录。"""
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
await QueryHistoryService.record_query(
|
|
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,
|
|
)
|
|
|
|
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: AsyncClient, admin_headers: dict):
|
|
"""测试按类型筛选。"""
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
await QueryHistoryService.record_query(
|
|
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 QueryHistoryService.record_query(
|
|
user_id=1, 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,
|
|
)
|
|
|
|
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: AsyncClient, admin_headers: dict):
|
|
"""测试收藏API。"""
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
entry = await QueryHistoryService.record_query(
|
|
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,
|
|
)
|
|
|
|
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: AsyncClient, admin_headers: dict):
|
|
"""测试收藏列表API。"""
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
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: AsyncClient, admin_headers: dict):
|
|
"""测试统计API。"""
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
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: AsyncClient, admin_headers: dict):
|
|
"""测试快捷查询列表API。"""
|
|
QueryHistoryService._quick_queries = []
|
|
QueryHistoryService._quick_counter = 0
|
|
QueryHistoryService.seed_default_quick_queries()
|
|
|
|
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: AsyncClient, admin_headers: dict):
|
|
"""测试创建快捷查询API。"""
|
|
QueryHistoryService._quick_queries = []
|
|
QueryHistoryService._quick_counter = 0
|
|
|
|
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: AsyncClient, admin_headers: dict):
|
|
"""测试快捷查询筛选API。"""
|
|
QueryHistoryService._quick_queries = []
|
|
QueryHistoryService._quick_counter = 0
|
|
QueryHistoryService.seed_default_quick_queries()
|
|
|
|
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: AsyncClient, guest_headers: dict):
|
|
"""测试访客可以查看自己的历史（但无权执行查询）。"""
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
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: AsyncClient, admin_headers: dict):
|
|
"""测试分页参数。"""
|
|
QueryHistoryService._history = []
|
|
QueryHistoryService._history_counter = 0
|
|
QueryHistoryService._favorites = set()
|
|
|
|
for i in range(15):
|
|
await QueryHistoryService.record_query(
|
|
user_id=1, 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,
|
|
)
|
|
|
|
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
|