拆分系统监测模块并统一运维化文案

This commit is contained in:
Cheng Zhou
2026-06-11 11:06:59 +08:00
parent b7484c8e01
commit 1886765db8
21 changed files with 222 additions and 143 deletions
+2 -5
View File
@@ -14,7 +14,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import func, select, desc
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.deps import get_current_user, get_db_session, is_system_admin, list_active_pm_study_ids
from app.core.deps import get_current_user, get_db_session, is_system_admin
from app.core.permission_monitor import get_permission_monitor
from app.models.permission_access_log import PermissionAccessLog
from app.models.permission_metric_snapshot import PermissionMetricSnapshot
@@ -39,10 +39,7 @@ class MonitoringScope:
async def resolve_monitoring_scope(db: AsyncSession, current_user) -> MonitoringScope:
if is_system_admin(current_user):
return MonitoringScope(is_admin=True, study_ids=set())
return MonitoringScope(
is_admin=False,
study_ids=await list_active_pm_study_ids(db, current_user.id),
)
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="权限不足")
def _apply_monitoring_scope_to_log_query(query, scope: MonitoringScope):
+12 -12
View File
@@ -969,54 +969,54 @@ SYSTEM_PERMISSIONS: dict[str, dict] = {
"description": "管理权限模板",
"roles": ["ADMIN"],
},
# 权限监控
# 系统监测
"system:monitoring:metrics": {
"module": "system_monitoring",
"action": "read",
"description": "查看权限监控指标",
"roles": ["ADMIN", "PM"],
"description": "查看系统监测指标",
"roles": ["ADMIN"],
},
"system:monitoring:cache_stats": {
"module": "system_monitoring",
"action": "read",
"description": "查看缓存统计",
"roles": ["ADMIN", "PM"],
"roles": ["ADMIN"],
},
"system:monitoring:alerts": {
"module": "system_monitoring",
"action": "read",
"description": "查看权限告警",
"roles": ["ADMIN", "PM"],
"roles": ["ADMIN"],
},
"system:monitoring:health": {
"module": "system_monitoring",
"action": "read",
"description": "查看权限系统健康状态",
"roles": ["ADMIN", "PM"],
"roles": ["ADMIN"],
},
"system:monitoring:access_logs": {
"module": "system_monitoring",
"action": "read",
"description": "查看权限访问日志",
"roles": ["ADMIN", "PM"],
"roles": ["ADMIN"],
},
"system:monitoring:trends": {
"module": "system_monitoring",
"action": "read",
"description": "查看权限趋势数据",
"roles": ["ADMIN", "PM"],
"roles": ["ADMIN"],
},
"system:monitoring:reset_metrics": {
"module": "system_monitoring",
"action": "update",
"description": "重置监控指标",
"roles": ["ADMIN", "PM"],
"description": "重置系统监测指标",
"roles": ["ADMIN"],
},
"system:monitoring:clear_alerts": {
"module": "system_monitoring",
"action": "update",
"description": "清除告警",
"roles": ["ADMIN", "PM"],
"roles": ["ADMIN"],
},
"system:monitoring:security_logs": {
"module": "system_monitoring",
@@ -1044,6 +1044,6 @@ SYSTEM_MODULE_LABELS: dict[str, str] = {
"system_users": "账号管理",
"system_projects": "项目管理",
"system_permissions": "权限管理",
"system_monitoring": "权限监控",
"system_monitoring": "系统监测",
"system_audit": "审计日志",
}
@@ -3,10 +3,12 @@
import uuid
import pytest
from fastapi import HTTPException
from sqlalchemy import text
from app.core.permission_monitor import set_permission_monitor, PermissionMonitor
from app.api.v1 import permission_monitoring
from app.api.v1.system_permissions import list_system_permissions
class FakeIpInfo:
@@ -22,6 +24,35 @@ class AdminUserStub:
is_admin = True
class ProjectPmUserStub:
id = uuid.uuid4()
is_admin = False
@pytest.mark.asyncio
async def test_resolve_monitoring_scope_rejects_project_pm(db_session):
"""系统监测模块仅允许系统管理员访问,项目 PM 不再具备监测范围。"""
with pytest.raises(HTTPException) as exc_info:
await permission_monitoring.resolve_monitoring_scope(db_session, ProjectPmUserStub())
assert exc_info.value.status_code == 403
@pytest.mark.asyncio
async def test_system_permission_monitoring_definitions_are_admin_only():
"""系统级权限定义中的系统监测模块不应再标注 PM 角色。"""
data = await list_system_permissions()
monitoring_items = [
item for item in data["permissions"]
if item["module"] == "system_monitoring"
]
assert monitoring_items
assert data["module_labels"]["system_monitoring"] == "系统监测"
assert all(item["roles"] == ["ADMIN"] for item in monitoring_items)
assert all("PM" not in item["roles"] for item in monitoring_items)
async def _seed_permission_log(db_session, study_id: uuid.UUID, user_id: uuid.UUID, *, allowed: bool, elapsed_ms: float) -> None:
study_exists = (
await db_session.execute(text("SELECT id FROM studies WHERE id = :id"), {"id": str(study_id)})