logan
305bbb153c
新增测试文件: - test_user_endpoint.py (18 tests): 用户CRUD/列表过滤/权限控制/自删保护 - test_pool_endpoint.py (12 tests): 连接池状态/重置/概览/mock外部依赖 - test_schema_drift_endpoint.py (13 tests): Schema快照捕获/漂移检测/历史查询 覆盖率提升: - user端点: 0→18 tests (CRUD+权限+重复用户名+自删保护) - pool端点: 0→12 tests (stats/reset/overview+角色访问控制) - schema_drift端点: 0→13 tests (capture/check/history+dict序列化兼容) 修复: run.sh启动脚本同步
390 lines
14 KiB
Python
390 lines
14 KiB
Python
"""P1 操作级权限测试: InstanceGroup.permissions 字段生效
|
||
|
||
测试矩阵:
|
||
- admin 不受 permissions 限制
|
||
- 组没配置 permissions → 允许所有（向后兼容）
|
||
- 组配置了 ["query:select"] → 只有 select 允许，ticket:create 被拒绝
|
||
- 用户级覆盖权限优先于组默认
|
||
- 公开实例不受 permissions 限制
|
||
- 集成测试: query/ticket 端点权限检查
|
||
"""
|
||
import json
|
||
import sys
|
||
import pytest
|
||
from contextlib import asynccontextmanager
|
||
from fastapi import HTTPException
|
||
from unittest.mock import patch, AsyncMock, MagicMock
|
||
|
||
from app.models.instance import Instance
|
||
from app.models.instance_group import InstanceGroup, user_groups
|
||
from app.core.security import create_access_token
|
||
|
||
|
||
# ── Fixtures ──────────────────────────────────────────────
|
||
|
||
@pytest.fixture
|
||
async def group_with_permissions(db):
|
||
"""创建带 permissions 配置的权限组"""
|
||
group = InstanceGroup(
|
||
name="只读组",
|
||
db_type="mysql",
|
||
permissions=json.dumps(["query:select", "instance:detail"]),
|
||
)
|
||
db.add(group)
|
||
await db.flush()
|
||
|
||
instance = Instance(
|
||
name="生产MySQL", db_type="mysql", host="10.0.0.1", port=3306,
|
||
username="root", password="pw", database="prod_db",
|
||
status="disconnected", group_id=group.id,
|
||
)
|
||
db.add(instance)
|
||
await db.commit()
|
||
await db.refresh(group)
|
||
await db.refresh(instance)
|
||
return group, instance
|
||
|
||
|
||
@pytest.fixture
|
||
async def group_no_permissions(db):
|
||
"""创建没有 permissions 配置的权限组（向后兼容）"""
|
||
group = InstanceGroup(name="默认组", db_type="mysql", permissions=None)
|
||
db.add(group)
|
||
await db.flush()
|
||
|
||
instance = Instance(
|
||
name="测试MySQL", db_type="mysql", host="127.0.0.1", port=3306,
|
||
username="root", password="pw", database="test_db",
|
||
status="disconnected", group_id=group.id,
|
||
)
|
||
db.add(instance)
|
||
await db.commit()
|
||
await db.refresh(group)
|
||
await db.refresh(instance)
|
||
return group, instance
|
||
|
||
|
||
# ── 单元测试: check_instance_permission ──────────────────
|
||
|
||
class TestCheckInstancePermission:
|
||
"""check_instance_permission 核心逻辑测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_admin_bypasses_permission_check(
|
||
self, db, admin_user, group_with_permissions
|
||
):
|
||
"""admin 不受 permissions 限制"""
|
||
from app.core.permissions import check_instance_permission
|
||
group, instance = group_with_permissions
|
||
# ticket:create 不在组 permissions 中，但 admin 应该通过
|
||
await check_instance_permission(admin_user, instance, "ticket:create", db)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_public_instance_allows_all(self, db, developer_user):
|
||
"""公开实例不受 permissions 限制"""
|
||
from app.core.permissions import check_instance_permission
|
||
instance = Instance(
|
||
name="公开", db_type="mysql", host="127.0.0.1", port=3306,
|
||
username="root", password="pw", database="pub_db",
|
||
status="disconnected", group_id=None,
|
||
)
|
||
db.add(instance)
|
||
await db.commit()
|
||
await check_instance_permission(developer_user, instance, "ticket:create", db)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_allowed_action_passes(
|
||
self, db, developer_user, group_with_permissions
|
||
):
|
||
"""用户在组内，操作在 permissions 列表中 → 通过"""
|
||
from app.core.permissions import check_instance_permission
|
||
group, instance = group_with_permissions
|
||
await db.execute(
|
||
user_groups.insert().values(
|
||
user_id=developer_user.id, group_id=group.id
|
||
)
|
||
)
|
||
await db.commit()
|
||
# query:select 在组 permissions 中
|
||
await check_instance_permission(developer_user, instance, "query:select", db)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_denied_action_raises_403(
|
||
self, db, developer_user, group_with_permissions
|
||
):
|
||
"""用户在组内，操作不在 permissions 列表中 → 403"""
|
||
from app.core.permissions import check_instance_permission
|
||
group, instance = group_with_permissions
|
||
await db.execute(
|
||
user_groups.insert().values(
|
||
user_id=developer_user.id, group_id=group.id
|
||
)
|
||
)
|
||
await db.commit()
|
||
# ticket:create 不在组 permissions ["query:select", "instance:detail"] 中
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
await check_instance_permission(
|
||
developer_user, instance, "ticket:create", db
|
||
)
|
||
assert exc_info.value.status_code == 403
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_no_permissions_config_allows_all(
|
||
self, db, developer_user, group_no_permissions
|
||
):
|
||
"""组没有配置 permissions → 允许所有操作（向后兼容）"""
|
||
from app.core.permissions import check_instance_permission
|
||
group, instance = group_no_permissions
|
||
await db.execute(
|
||
user_groups.insert().values(
|
||
user_id=developer_user.id, group_id=group.id
|
||
)
|
||
)
|
||
await db.commit()
|
||
# ticket:create 应该被允许，因为组没有配置限制
|
||
await check_instance_permission(developer_user, instance, "ticket:create", db)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_user_override_permissions(
|
||
self, db, developer_user, group_with_permissions
|
||
):
|
||
"""用户级覆盖权限优先于组默认"""
|
||
from app.core.permissions import check_instance_permission
|
||
group, instance = group_with_permissions
|
||
# 组 permissions = ["query:select", "instance:detail"]
|
||
# 用户覆盖为 ["query:select"] — 只给查询权限
|
||
await db.execute(
|
||
user_groups.insert().values(
|
||
user_id=developer_user.id,
|
||
group_id=group.id,
|
||
permissions=json.dumps(["query:select"]),
|
||
)
|
||
)
|
||
await db.commit()
|
||
|
||
# query:select 通过（用户级覆盖包含）
|
||
await check_instance_permission(developer_user, instance, "query:select", db)
|
||
|
||
# instance:detail 被拒绝（组默认有，但用户覆盖只给了 query:select）
|
||
with pytest.raises(HTTPException) as exc_info:
|
||
await check_instance_permission(
|
||
developer_user, instance, "instance:detail", db
|
||
)
|
||
assert exc_info.value.status_code == 403
|
||
|
||
|
||
# ── 单元测试: _get_effective_permissions ──────────────────
|
||
|
||
class TestGetEffectivePermissions:
|
||
"""_get_effective_permissions 权限解析逻辑"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_all_when_no_config(
|
||
self, db, developer_user, group_no_permissions
|
||
):
|
||
"""无权限配置时返回全部权限"""
|
||
from app.core.permissions import _get_effective_permissions, PERMISSION_ACTIONS
|
||
group, instance = group_no_permissions
|
||
await db.execute(
|
||
user_groups.insert().values(
|
||
user_id=developer_user.id, group_id=group.id
|
||
)
|
||
)
|
||
await db.commit()
|
||
result = await _get_effective_permissions(developer_user, instance, db)
|
||
assert result == PERMISSION_ACTIONS
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_group_permissions(
|
||
self, db, developer_user, group_with_permissions
|
||
):
|
||
"""有组权限配置时返回组权限"""
|
||
from app.core.permissions import _get_effective_permissions
|
||
group, instance = group_with_permissions
|
||
await db.execute(
|
||
user_groups.insert().values(
|
||
user_id=developer_user.id, group_id=group.id
|
||
)
|
||
)
|
||
await db.commit()
|
||
result = await _get_effective_permissions(developer_user, instance, db)
|
||
assert result == {"query:select", "instance:detail"}
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_user_override_takes_precedence(
|
||
self, db, developer_user, group_with_permissions
|
||
):
|
||
"""用户级覆盖优先于组默认"""
|
||
from app.core.permissions import _get_effective_permissions
|
||
group, instance = group_with_permissions
|
||
await db.execute(
|
||
user_groups.insert().values(
|
||
user_id=developer_user.id,
|
||
group_id=group.id,
|
||
permissions=json.dumps(["query:select"]),
|
||
)
|
||
)
|
||
await db.commit()
|
||
result = await _get_effective_permissions(developer_user, instance, db)
|
||
assert result == {"query:select"}
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_admin_not_in_group(
|
||
self, db, admin_user, group_with_permissions
|
||
):
|
||
"""admin 不在任何组，但 check_instance_permission 不调用此函数（admin bypass）"""
|
||
# admin 在 check_instance_permission 中直接 return，不会调用 _get_effective_permissions
|
||
pass
|
||
|
||
|
||
# ── 集成测试: API 端点权限检查 ────────────────────────────
|
||
|
||
class TestPermissionEndpointIntegration:
|
||
"""权限检查在实际 API 端点中的集成"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_query_with_permission(
|
||
self, client, db, developer_user, group_with_permissions
|
||
):
|
||
"""有 query:select 权限 → 查询不被 403（mock DB连接避免挂起）"""
|
||
group, instance = group_with_permissions
|
||
await db.execute(
|
||
user_groups.insert().values(
|
||
user_id=developer_user.id, group_id=group.id
|
||
)
|
||
)
|
||
await db.commit()
|
||
token = create_access_token(
|
||
data={"sub": str(developer_user.id), "role": developer_user.role}
|
||
)
|
||
|
||
# Mock aiomysql 防止真实连接挂起（见 skill #20, #37）
|
||
mock_cursor = AsyncMock()
|
||
mock_cursor.description = [("1",)]
|
||
mock_cursor.fetchall.return_value = [{"1": 1}]
|
||
mock_cursor.close = AsyncMock()
|
||
|
||
@asynccontextmanager
|
||
async def mock_cursor_ctx(*a, **kw):
|
||
yield mock_cursor
|
||
|
||
mock_conn = AsyncMock()
|
||
mock_conn.cursor = mock_cursor_ctx
|
||
mock_conn.commit = AsyncMock()
|
||
mock_conn.rollback = AsyncMock()
|
||
mock_conn.close = AsyncMock()
|
||
|
||
mock_aiomysql = MagicMock()
|
||
mock_aiomysql.connect = AsyncMock(return_value=mock_conn)
|
||
mock_aiomysql.DictCursor = MagicMock()
|
||
|
||
with patch.dict(sys.modules, {"aiomysql": mock_aiomysql}):
|
||
resp = await client.post(
|
||
"/api/v1/query",
|
||
json={
|
||
"instance_id": instance.id,
|
||
"database": "test",
|
||
"sql": "SELECT 1",
|
||
},
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
)
|
||
# 不应是 403（权限通过，mock 返回成功）
|
||
assert resp.status_code != 403
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ticket_create_without_permission(
|
||
self, client, db, developer_user, group_with_permissions
|
||
):
|
||
"""没有 ticket:create 权限 → 创建工单 403"""
|
||
group, instance = group_with_permissions
|
||
await db.execute(
|
||
user_groups.insert().values(
|
||
user_id=developer_user.id, group_id=group.id
|
||
)
|
||
)
|
||
await db.commit()
|
||
token = create_access_token(
|
||
data={"sub": str(developer_user.id), "role": developer_user.role}
|
||
)
|
||
resp = await client.post(
|
||
"/api/v1/tickets",
|
||
json={
|
||
"title": "测试工单标题",
|
||
"db_type": "mysql",
|
||
"instance_id": instance.id,
|
||
"target_database": "test",
|
||
"sql_content": "ALTER TABLE t ADD COLUMN c INT",
|
||
"reason": "测试创建工单权限",
|
||
},
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
)
|
||
assert resp.status_code == 403
|
||
assert "ticket:create" in resp.json()["detail"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ticket_create_with_permission(
|
||
self, client, db, developer_user, group_no_permissions
|
||
):
|
||
"""组没配置 permissions → 创建工单不被 403（向后兼容）"""
|
||
group, instance = group_no_permissions
|
||
await db.execute(
|
||
user_groups.insert().values(
|
||
user_id=developer_user.id, group_id=group.id
|
||
)
|
||
)
|
||
await db.commit()
|
||
token = create_access_token(
|
||
data={"sub": str(developer_user.id), "role": developer_user.role}
|
||
)
|
||
resp = await client.post(
|
||
"/api/v1/tickets",
|
||
json={
|
||
"title": "测试工单标题",
|
||
"db_type": "mysql",
|
||
"instance_id": instance.id,
|
||
"target_database": "test",
|
||
"sql_content": "ALTER TABLE t ADD COLUMN c INT",
|
||
"reason": "测试创建工单权限",
|
||
},
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
)
|
||
# 不应是 403（可能 400 因为数据校验，但权限通过）
|
||
assert resp.status_code != 403
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_query_denied_for_group_without_select(
|
||
self, client, db, developer_user
|
||
):
|
||
"""组 permissions 只有 ticket:create → query 被拒绝"""
|
||
group = InstanceGroup(
|
||
name="工单组",
|
||
db_type="mysql",
|
||
permissions=json.dumps(["ticket:create"]),
|
||
)
|
||
db.add(group)
|
||
await db.flush()
|
||
|
||
instance = Instance(
|
||
name="工单MySQL", db_type="mysql", host="10.0.0.1", port=3306,
|
||
username="root", password="pw", database="prod_db",
|
||
status="disconnected", group_id=group.id,
|
||
)
|
||
db.add(instance)
|
||
await db.execute(
|
||
user_groups.insert().values(
|
||
user_id=developer_user.id, group_id=group.id
|
||
)
|
||
)
|
||
await db.commit()
|
||
|
||
token = create_access_token(
|
||
data={"sub": str(developer_user.id), "role": developer_user.role}
|
||
)
|
||
resp = await client.post(
|
||
"/api/v1/query",
|
||
json={"instance_id": instance.id, "database": "test", "sql": "SELECT 1"},
|
||
headers={"Authorization": f"Bearer {token}"},
|
||
)
|
||
assert resp.status_code == 403
|
||
assert "query:select" in resp.json()["detail"]
|