feat(perm): 项目 PM 共管接口权限与监控并细化 PM 写权限边界

- deps 新增 list_active_pm_study_ids、is_active_project_pm 与
  require_admin_or_any_project_pm 依赖，用于把 PM 项目范围带进鉴权与
  监控。get_cra_site_scope 内部延后导入 site CRUD，避免循环依赖。
- system_permissions API 改用 PM/ADMIN 双角色入口，permissions/monitoring
  系统级权限新增 PM 配额，并细化访问日志、告警、监控指标的可见范围。
- members API 调整：项目 PM 仅可管理低于 PM 的项目角色，禁止互相
  改写或授予 PM。
- api_permissions API 增加 GET /api-permissions/me，返回当前用户在该
  项目的有效权限矩阵；保存权限矩阵时校验 PM 行为不被篡改。
- core/api_permissions：新增立项配置接口键、PM 默认拥有的监控/权限
  系统级条目，并在权限元信息中标注 PM 共享角色。
- core/project_permissions：role_has_api_permission 命中默认角色矩阵；
  replace_api_endpoint_permissions 改为部分更新且永不持久化 ADMIN/PM。
- studies setup-config 各端点改用接口级权限装饰器，与新的 setup_config
  权限键对齐。permission_monitor 新增 get_metrics 摘要供 PM 视图调用。
- 测试：新增 test_admin_pm_permissions 覆盖 PM 系统级权限、监控范围和
  成员管理边界；conftest 兼容 SA_UUID 列；权限相关用例同步移除已失效
  的 module_permission 链路。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/tests/conftest.py

@@ -8,7 +8,7 @@ from typing import AsyncGenerator
 
 import pytest
 import pytest_asyncio
-from sqlalchemy import text, event, String, TypeDecorator
+from sqlalchemy import UUID as SA_UUID, text, event, String, TypeDecorator
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.dialects.postgresql import UUID as PG_UUID
 
@@ -90,7 +90,7 @@ async def _create_test_engine():
     # Replace PostgreSQL UUID type with custom GUID type for SQLite
     for table in Base.metadata.tables.values():
         for column in table.columns:
-            if isinstance(column.type, PG_UUID):
+            if isinstance(column.type, (PG_UUID, SA_UUID)):
                 column.type = GUID()
 
     # Create all tables

backend/tests/test_admin_pm_permissions.py

@@ -0,0 +1,260 @@
+import uuid
+from dataclasses import dataclass
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.v1.api_permissions import get_my_study_api_permissions, update_study_api_permissions
+from app.api.v1.members import update_member
+from app.api.v1.permission_monitoring import resolve_monitoring_scope
+from app.api.v1.system_permissions import list_system_permissions
+from app.core.deps import require_admin_or_any_project_pm
+from app.schemas.member import StudyMemberUpdate
+
+
+@dataclass
+class UserStub:
+    id: uuid.UUID
+    role: str
+
+
+async def _seed_user(db: AsyncSession, user_id: uuid.UUID, role: str = "PM") -> None:
+    await db.execute(
+        text(
+            """
+            INSERT INTO users (id, email, password_hash, full_name, role, clinical_department, status)
+            VALUES (:id, :email, :password_hash, :full_name, :role, :clinical_department, :status)
+            """
+        ),
+        {
+            "id": str(user_id),
+            "email": f"{user_id.hex}@example.com",
+            "password_hash": "hash",
+            "full_name": f"User {user_id.hex[:6]}",
+            "role": role,
+            "clinical_department": "Clinical",
+            "status": "ACTIVE",
+        },
+    )
+
+
+async def _seed_study(db: AsyncSession, study_id: uuid.UUID, code: str) -> None:
+    await db.execute(
+        text(
+            """
+            INSERT INTO studies (id, code, name, status, is_locked, visit_schedule, active_roles)
+            VALUES (:id, :code, :name, :status, :is_locked, :visit_schedule, :active_roles)
+            """
+        ),
+        {
+            "id": str(study_id),
+            "code": code,
+            "name": code,
+            "status": "ACTIVE",
+            "is_locked": False,
+            "visit_schedule": "[]",
+            "active_roles": "[]",
+        },
+    )
+
+
+async def _seed_member(db: AsyncSession, study_id: uuid.UUID, user_id: uuid.UUID, role: str, active: bool = True) -> None:
+    await db.execute(
+        text(
+            """
+            INSERT INTO study_members (id, study_id, user_id, role_in_study, is_active)
+            VALUES (:id, :study_id, :user_id, :role, :active)
+            """
+        ),
+        {
+            "id": str(uuid.uuid4()),
+            "study_id": str(study_id),
+            "user_id": str(user_id),
+            "role": role,
+            "active": active,
+        },
+    )
+
+
+async def _seed_member_return_id(
+    db: AsyncSession,
+    study_id: uuid.UUID,
+    user_id: uuid.UUID,
+    role: str,
+    active: bool = True,
+) -> uuid.UUID:
+    member_id = uuid.uuid4()
+    await db.execute(
+        text(
+            """
+            INSERT INTO study_members (id, study_id, user_id, role_in_study, is_active)
+            VALUES (:id, :study_id, :user_id, :role, :active)
+            """
+        ),
+        {
+            "id": str(member_id),
+            "study_id": str(study_id),
+            "user_id": str(user_id),
+            "role": role,
+            "active": active,
+        },
+    )
+    return member_id
+
+
+@pytest.mark.asyncio
+async def test_project_pm_can_view_system_permission_definitions(db_session: AsyncSession):
+    pm_id = uuid.uuid4()
+    study_id = uuid.uuid4()
+    await _seed_user(db_session, pm_id)
+    await _seed_study(db_session, study_id, "PM-SYSTEM-PERMS")
+    await _seed_member(db_session, study_id, pm_id, "PM")
+    await db_session.commit()
+
+    dependency = require_admin_or_any_project_pm()
+    await dependency(current_user=UserStub(id=pm_id, role="PM"), db=db_session)
+    data = await list_system_permissions()
+
+    assert data["permissions"]
+
+
+@pytest.mark.asyncio
+async def test_non_pm_cannot_view_system_permission_definitions(db_session: AsyncSession):
+    cra_id = uuid.uuid4()
+    study_id = uuid.uuid4()
+    await _seed_user(db_session, cra_id, role="CRA")
+    await _seed_study(db_session, study_id, "CRA-SYSTEM-PERMS")
+    await _seed_member(db_session, study_id, cra_id, "CRA")
+    await db_session.commit()
+
+    with pytest.raises(HTTPException) as exc_info:
+        dependency = require_admin_or_any_project_pm()
+        await dependency(current_user=UserStub(id=cra_id, role="CRA"), db=db_session)
+
+    assert exc_info.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_pm_permission_update_does_not_persist_admin_or_pm_overrides(db_session: AsyncSession):
+    study_id = uuid.uuid4()
+    admin_id = uuid.uuid4()
+    await _seed_study(db_session, study_id, "PM-PERM-SKIP")
+    await _seed_user(db_session, admin_id, role="ADMIN")
+    await db_session.commit()
+
+    result = await update_study_api_permissions(
+        study_id=study_id,
+        payload={
+            "ADMIN": {"subjects:delete": False},
+            "PM": {"subjects:delete": False},
+            "CRA": {"subjects:delete": True},
+        },
+        current_user=UserStub(id=admin_id, role="ADMIN"),
+        db=db_session,
+    )
+
+    rows = (
+        await db_session.execute(
+            text(
+                """
+                SELECT role, endpoint_key, allowed
+                FROM api_endpoint_permissions
+                WHERE study_id = :study_id
+                """
+            ),
+            {"study_id": str(study_id)},
+        )
+    ).all()
+    assert ("CRA", "subjects:delete", True) in rows
+    assert all(row.role not in {"ADMIN", "PM"} for row in rows)
+    assert "ADMIN" not in result
+    assert result["PM"]["subjects:delete"]["allowed"] is True
+
+
+@pytest.mark.asyncio
+async def test_project_member_can_read_own_effective_permissions(db_session: AsyncSession):
+    cra_id = uuid.uuid4()
+    study_id = uuid.uuid4()
+    await _seed_user(db_session, cra_id, role="CRA")
+    await _seed_study(db_session, study_id, "CRA-MY-PERMS")
+    await _seed_member(db_session, study_id, cra_id, "CRA")
+    await db_session.commit()
+
+    result = await get_my_study_api_permissions(
+        study_id=study_id,
+        current_user=UserStub(id=cra_id, role="CRA"),
+        db=db_session,
+    )
+
+    assert list(result) == ["CRA"]
+    assert result["CRA"]["sites:read"]["allowed"] is True
+    assert result["CRA"]["sites:update"]["allowed"] is False
+
+
+@pytest.mark.asyncio
+async def test_project_pm_monitoring_scope_is_limited_to_own_projects(db_session: AsyncSession):
+    pm_id = uuid.uuid4()
+    own_study_id = uuid.uuid4()
+    other_study_id = uuid.uuid4()
+    await _seed_user(db_session, pm_id)
+    await _seed_study(db_session, own_study_id, "PM-MONITOR-OWN")
+    await _seed_study(db_session, other_study_id, "PM-MONITOR-OTHER")
+    await _seed_member(db_session, own_study_id, pm_id, "PM")
+    await db_session.commit()
+
+    scope = await resolve_monitoring_scope(db_session, UserStub(id=pm_id, role="PM"))
+
+    assert scope.is_admin is False
+    assert scope.study_ids == {own_study_id}
+    assert scope.can_access_study(own_study_id)
+    assert not scope.can_access_study(other_study_id)
+
+
+@pytest.mark.asyncio
+async def test_project_pm_cannot_update_peer_pm_member(db_session: AsyncSession):
+    study_id = uuid.uuid4()
+    actor_id = uuid.uuid4()
+    peer_id = uuid.uuid4()
+    await _seed_study(db_session, study_id, "PM-PEER-MEMBER")
+    await _seed_user(db_session, actor_id)
+    await _seed_user(db_session, peer_id)
+    await _seed_member(db_session, study_id, actor_id, "PM")
+    peer_member_id = await _seed_member_return_id(db_session, study_id, peer_id, "PM")
+    await db_session.commit()
+
+    with pytest.raises(HTTPException) as exc_info:
+        await update_member(
+            study_id=study_id,
+            member_id=peer_member_id,
+            member_in=StudyMemberUpdate(role_in_study="CRA"),
+            current_user=UserStub(id=actor_id, role="PM"),
+            db=db_session,
+        )
+
+    assert exc_info.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_project_pm_cannot_grant_peer_pm_role(db_session: AsyncSession):
+    study_id = uuid.uuid4()
+    actor_id = uuid.uuid4()
+    cra_id = uuid.uuid4()
+    await _seed_study(db_session, study_id, "PM-GRANT-PM")
+    await _seed_user(db_session, actor_id)
+    await _seed_user(db_session, cra_id, role="CRA")
+    await _seed_member(db_session, study_id, actor_id, "PM")
+    cra_member_id = await _seed_member_return_id(db_session, study_id, cra_id, "CRA")
+    await db_session.commit()
+
+    with pytest.raises(HTTPException) as exc_info:
+        await update_member(
+            study_id=study_id,
+            member_id=cra_member_id,
+            member_in=StudyMemberUpdate(role_in_study="PM"),
+            current_user=UserStub(id=actor_id, role="PM"),
+            db=db_session,
+        )
+
+    assert exc_info.value.status_code == 403

backend/tests/test_api_permissions.py

@@ -64,6 +64,23 @@ async def test_default_matrix_covers_every_role_and_permission(db_session: Async
             assert matrix[role][endpoint_key]["allowed"] is (role in config["default_roles"])
 
 
+@pytest.mark.asyncio
+async def test_default_matrix_round_trips_to_backend_checks(db_session: AsyncSession):
+    """默认权限矩阵应与后端实际鉴权结果一致。"""
+    study_id = uuid.uuid4()
+
+    for role in PROJECT_PERMISSION_ROLES:
+        for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
+            allowed = await role_has_api_permission(
+                db_session,
+                study_id,
+                role,
+                endpoint_key,
+                check_prerequisites=False,
+            )
+            assert allowed is (role in config["default_roles"])
+
+
 @pytest.mark.asyncio
 async def test_full_permission_matrix_round_trips_to_backend_checks(db_session: AsyncSession):
     """逐一验证前端提交格式会落库，并被后端鉴权函数按相同结果读取。"""
@@ -73,17 +90,18 @@ async def test_full_permission_matrix_round_trips_to_backend_checks(db_session:
     set_permission_monitor(monitor)
 
     study_id = uuid.uuid4()
+    configurable_roles = [role for role in PROJECT_PERMISSION_ROLES if role != "PM"]
     payload = {
         role: {
             endpoint_key: index % 2 == role_index % 2
             for index, endpoint_key in enumerate(API_ENDPOINT_PERMISSIONS)
         }
-        for role_index, role in enumerate(PROJECT_PERMISSION_ROLES)
+        for role_index, role in enumerate(configurable_roles)
     }
 
     matrix = await replace_api_endpoint_permissions(db_session, study_id, payload)
 
-    for role in PROJECT_PERMISSION_ROLES:
+    for role in configurable_roles:
         for endpoint_key, expected in payload[role].items():
             assert matrix[role][endpoint_key]["allowed"] is expected
             allowed = await role_has_api_permission(

backend/tests/test_api_permissions_endpoints.py

@@ -237,6 +237,58 @@ async def test_replace_api_endpoint_permissions_partial_update(db_session: Async
     assert result["CRA"]["subjects:list"]["allowed"] is True
 
 
+@pytest.mark.asyncio
+async def test_replace_api_endpoint_permissions_preserves_unsubmitted_permissions_for_same_role(
+    db_session: AsyncSession,
+):
+    """保存角色的部分权限时，不应清空该角色未提交的权限项。"""
+    study_id = uuid.uuid4()
+
+    await replace_api_endpoint_permissions(
+        db_session,
+        study_id,
+        {
+            "CRA": {
+                "subjects:create": True,
+                "subjects:delete": True,
+            }
+        },
+    )
+
+    result = await replace_api_endpoint_permissions(
+        db_session,
+        study_id,
+        {"CRA": {"subjects:create": False}},
+    )
+
+    assert result["CRA"]["subjects:create"]["allowed"] is False
+    assert result["CRA"]["subjects:delete"]["allowed"] is True
+
+
+@pytest.mark.asyncio
+async def test_replace_api_endpoint_permissions_preserves_unsubmitted_roles(db_session: AsyncSession):
+    """保存单个角色权限时，不应清空其他角色的已配置权限。"""
+    study_id = uuid.uuid4()
+
+    await replace_api_endpoint_permissions(
+        db_session,
+        study_id,
+        {
+            "CRA": {"subjects:create": True},
+            "PV": {"subjects:create": True},
+        },
+    )
+
+    result = await replace_api_endpoint_permissions(
+        db_session,
+        study_id,
+        {"CRA": {"subjects:create": False}},
+    )
+
+    assert result["CRA"]["subjects:create"]["allowed"] is False
+    assert result["PV"]["subjects:create"]["allowed"] is True
+
+
 @pytest.mark.asyncio
 async def test_get_api_endpoint_permissions_structure(db_session: AsyncSession):
     """测试权限矩阵的结构"""

backend/tests/test_api_permissions_prerequisites.py

@@ -2,22 +2,20 @@
 
 import pytest
 import uuid
-from httpx import AsyncClient
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.models.study import Study
-from app.models.user import User
-from app.models.study_member import StudyMember
+from app.api.v1.api_permissions import (
+    check_operation_prerequisites,
+    list_api_operations,
+    list_operation_prerequisites,
+)
 from app.models.api_endpoint_permission import ApiEndpointPermission
 
 
 @pytest.mark.asyncio
-async def test_list_operations_with_prerequisites(client: AsyncClient, db_session: AsyncSession):
+async def test_list_operations_with_prerequisites():
     """测试获取所有权限操作及其前置权限"""
-    response = await client.get("/api-permissions/operations")
-
-    assert response.status_code == 200
-    data = response.json()
+    data = await list_api_operations()
     assert "operations" in data
 
     # 验证返回的操作包含前置权限字段
@@ -35,12 +33,9 @@ async def test_list_operations_with_prerequisites(client: AsyncClient, db_sessio
 
 
 @pytest.mark.asyncio
-async def test_list_operation_prerequisites(client: AsyncClient):
+async def test_list_operation_prerequisites_endpoint():
     """测试获取所有操作的前置权限依赖"""
-    response = await client.get("/api-permissions/operations/prerequisites")
-
-    assert response.status_code == 200
-    data = response.json()
+    data = await list_operation_prerequisites()
     assert "prerequisites" in data
 
     prerequisites = data["prerequisites"]
@@ -57,9 +52,10 @@ async def test_list_operation_prerequisites(client: AsyncClient):
 
 @pytest.mark.asyncio
 async def test_check_operation_prerequisites_satisfied(
-    client: AsyncClient, db_session: AsyncSession, study_id: uuid.UUID
+    db_session: AsyncSession
 ):
     """测试检查操作前置权限 - 满足"""
+    study_id = uuid.uuid4()
     # 创建权限：主权限 + 前置权限都允许
     main_perm = ApiEndpointPermission(
         study_id=study_id,
@@ -77,13 +73,13 @@ async def test_check_operation_prerequisites_satisfied(
     db_session.add(prereq_perm)
     await db_session.commit()
 
-    response = await client.get(
-        f"/api-permissions/subjects:create/prerequisites",
-        params={"study_id": str(study_id), "role": "CRA"}
+    data = await check_operation_prerequisites(
+        study_id=study_id,
+        endpoint_key="subjects:create",
+        role="CRA",
+        db=db_session,
     )
 
-    assert response.status_code == 200
-    data = response.json()
     assert data["endpoint_key"] == "subjects:create"
     assert data["role"] == "CRA"
     assert data["has_main_permission"] is True
@@ -93,9 +89,10 @@ async def test_check_operation_prerequisites_satisfied(
 
 @pytest.mark.asyncio
 async def test_check_operation_prerequisites_missing(
-    client: AsyncClient, db_session: AsyncSession, study_id: uuid.UUID
+    db_session: AsyncSession
 ):
     """测试检查操作前置权限 - 缺失"""
+    study_id = uuid.uuid4()
     # 创建权限：主权限允许，前置权限不允许
     main_perm = ApiEndpointPermission(
         study_id=study_id,
@@ -113,13 +110,13 @@ async def test_check_operation_prerequisites_missing(
     db_session.add(prereq_perm)
     await db_session.commit()
 
-    response = await client.get(
-        f"/api-permissions/subjects:create/prerequisites",
-        params={"study_id": str(study_id), "role": "CRA"}
+    data = await check_operation_prerequisites(
+        study_id=study_id,
+        endpoint_key="subjects:create",
+        role="CRA",
+        db=db_session,
     )
 
-    assert response.status_code == 200
-    data = response.json()
     assert data["endpoint_key"] == "subjects:create"
     assert data["role"] == "CRA"
     assert data["has_main_permission"] is True
@@ -129,9 +126,10 @@ async def test_check_operation_prerequisites_missing(
 
 @pytest.mark.asyncio
 async def test_check_operation_prerequisites_multiple_missing(
-    client: AsyncClient, db_session: AsyncSession, study_id: uuid.UUID
+    db_session: AsyncSession
 ):
     """测试检查操作前置权限 - 多个缺失"""
+    study_id = uuid.uuid4()
     # 创建权限：主权限允许，两个前置权限都不允许
     main_perm = ApiEndpointPermission(
         study_id=study_id,
@@ -156,13 +154,13 @@ async def test_check_operation_prerequisites_multiple_missing(
     db_session.add(prereq2)
     await db_session.commit()
 
-    response = await client.get(
-        f"/api-permissions/visits:create/prerequisites",
-        params={"study_id": str(study_id), "role": "CRA"}
+    data = await check_operation_prerequisites(
+        study_id=study_id,
+        endpoint_key="visits:create",
+        role="CRA",
+        db=db_session,
     )
 
-    assert response.status_code == 200
-    data = response.json()
     assert data["endpoint_key"] == "visits:create"
     assert data["role"] == "CRA"
     assert data["has_main_permission"] is True
@@ -171,17 +169,15 @@ async def test_check_operation_prerequisites_multiple_missing(
 
 
 @pytest.mark.asyncio
-async def test_check_operation_prerequisites_admin(
-    client: AsyncClient, study_id: uuid.UUID
-):
+async def test_check_operation_prerequisites_admin():
     """测试检查操作前置权限 - ADMIN角色"""
-    response = await client.get(
-        f"/api-permissions/subjects:create/prerequisites",
-        params={"study_id": str(study_id), "role": "ADMIN"}
+    data = await check_operation_prerequisites(
+        study_id=uuid.uuid4(),
+        endpoint_key="subjects:create",
+        role="ADMIN",
+        db=None,
     )
 
-    assert response.status_code == 200
-    data = response.json()
     assert data["role"] == "ADMIN"
     assert data["has_main_permission"] is True
     assert data["missing_prerequisites"] == []
@@ -190,9 +186,10 @@ async def test_check_operation_prerequisites_admin(
 
 @pytest.mark.asyncio
 async def test_check_operation_prerequisites_no_main_permission(
-    client: AsyncClient, db_session: AsyncSession, study_id: uuid.UUID
+    db_session: AsyncSession
 ):
     """测试检查操作前置权限 - 没有主权限"""
+    study_id = uuid.uuid4()
     # 创建权限：主权限不允许
     main_perm = ApiEndpointPermission(
         study_id=study_id,
@@ -203,13 +200,13 @@ async def test_check_operation_prerequisites_no_main_permission(
     db_session.add(main_perm)
     await db_session.commit()
 
-    response = await client.get(
-        f"/api-permissions/subjects:create/prerequisites",
-        params={"study_id": str(study_id), "role": "CRA"}
+    data = await check_operation_prerequisites(
+        study_id=study_id,
+        endpoint_key="subjects:create",
+        role="CRA",
+        db=db_session,
     )
 
-    assert response.status_code == 200
-    data = response.json()
     assert data["endpoint_key"] == "subjects:create"
     assert data["role"] == "CRA"
     assert data["has_main_permission"] is False

backend/tests/test_app_startup.py

@@ -10,6 +10,10 @@ from app import main
 class _DummyTask:
     def __init__(self):
         self.awaited = False
+        self.cancelled = False
+
+    def cancel(self):
+        self.cancelled = True
 
     def __await__(self):
         async def _wait():

backend/tests/test_permission_monitoring.py

@@ -1,66 +1,22 @@
-"""监控测试：权限系统监控功能验证
-
-测试权限系统的监控功能，包括：
-- 指标收集
-- 告警生成
-- 健康检查
-"""
+"""监控测试：权限系统内存监控职责验证。"""
 
 import time
-import uuid
 
 import pytest
 
 from app.core.permission_monitor import (
-    PermissionMonitor,
-    PermissionCheckMetrics,
     CacheMetrics,
+    PermissionMonitor,
     get_permission_monitor,
     set_permission_monitor,
-    evaluate_permission_system_health,
 )
 
 
-@pytest.mark.asyncio
-async def test_permission_check_metrics():
-    """测试权限检查指标"""
-    monitor = PermissionMonitor()
-
-    # 记录权限检查
-    monitor.record_permission_check(allowed=True, elapsed_time=0.005)
-    monitor.record_permission_check(allowed=False, elapsed_time=0.003)
-    monitor.record_permission_check(allowed=True, elapsed_time=0.004)
-
-    metrics = monitor.metrics.check_metrics
-    assert metrics.total_checks == 3
-    assert metrics.allowed_checks == 2
-    assert metrics.denied_checks == 1
-    assert metrics.allow_rate == pytest.approx(66.67, 0.1)
-    assert metrics.deny_rate == pytest.approx(33.33, 0.1)
-
-
-@pytest.mark.asyncio
-async def test_permission_check_timing():
-    """测试权限检查耗时统计"""
-    monitor = PermissionMonitor()
-
-    # 记录不同耗时的权限检查
-    monitor.record_permission_check(allowed=True, elapsed_time=0.001)
-    monitor.record_permission_check(allowed=True, elapsed_time=0.005)
-    monitor.record_permission_check(allowed=True, elapsed_time=0.003)
-
-    metrics = monitor.metrics.check_metrics
-    assert metrics.min_time == pytest.approx(0.001, 0.0001)
-    assert metrics.max_time == pytest.approx(0.005, 0.0001)
-    assert metrics.avg_time == pytest.approx(0.003, 0.0001)
-
-
 @pytest.mark.asyncio
 async def test_cache_metrics():
     """测试缓存指标"""
     monitor = PermissionMonitor()
 
-    # 记录缓存访问
     monitor.record_cache_hit()
     monitor.record_cache_hit()
     monitor.record_cache_miss()
@@ -79,41 +35,35 @@ async def test_cache_invalidation_tracking():
     """测试缓存失效追踪"""
     monitor = PermissionMonitor()
 
-    # 记录缓存失效
     monitor.record_cache_invalidation()
     monitor.record_cache_invalidation()
     monitor.record_cache_invalidation()
 
-    metrics = monitor.metrics.cache_metrics
-    assert metrics.cache_invalidations == 3
+    assert monitor.metrics.cache_metrics.cache_invalidations == 3
 
 
 @pytest.mark.asyncio
-async def test_error_tracking():
-    """测试错误追踪"""
+async def test_slow_check_alert_generation():
+    """测试慢权限检查告警生成"""
     monitor = PermissionMonitor()
 
-    # 记录权限检查错误
-    error = ValueError("test error")
-    monitor.record_permission_check(allowed=False, elapsed_time=0.005, error=error)
-    monitor.record_permission_check(allowed=True, elapsed_time=0.003)
-
-    metrics = monitor.metrics.check_metrics
-    assert metrics.errors == 1
-    assert metrics.error_rate == pytest.approx(50.0, 0.1)
-
-
-@pytest.mark.asyncio
-async def test_alert_generation():
-    """测试告警生成"""
-    monitor = PermissionMonitor()
-
-    # 记录慢速权限检查（应该生成告警）
-    monitor.record_permission_check(allowed=True, elapsed_time=0.1)
+    monitor.record_slow_check_alert(100)
 
     alerts = monitor.get_alerts()
-    assert len(alerts) > 0
+    assert len(alerts) == 1
+    assert alerts[0]["level"] == "warning"
     assert alerts[0]["type"] == "slow_permission_check"
+    assert alerts[0]["data"]["elapsed_ms"] == 100
+
+
+@pytest.mark.asyncio
+async def test_slow_check_alert_ignores_fast_checks():
+    """未超过阈值的权限检查不应生成告警"""
+    monitor = PermissionMonitor()
+
+    monitor.record_slow_check_alert(10)
+
+    assert monitor.get_alerts() == []
 
 
 @pytest.mark.asyncio
@@ -121,12 +71,11 @@ async def test_error_alert_generation():
     """测试错误告警生成"""
     monitor = PermissionMonitor()
 
-    # 记录权限检查错误（应该生成告警）
-    error = ValueError("test error")
-    monitor.record_permission_check(allowed=False, elapsed_time=0.005, error=error)
+    monitor.record_error_alert(ValueError("test error"))
 
     alerts = monitor.get_alerts()
-    assert len(alerts) > 0
+    assert len(alerts) == 1
+    assert alerts[0]["level"] == "error"
     assert alerts[0]["type"] == "permission_check_error"
 
 
@@ -135,20 +84,16 @@ async def test_alert_filtering():
     """测试告警过滤"""
     monitor = PermissionMonitor()
 
-    # 生成不同级别的告警
-    monitor.record_permission_check(allowed=True, elapsed_time=0.1)  # warning
-    error = ValueError("test error")
-    monitor.record_permission_check(allowed=False, elapsed_time=0.005, error=error)  # error
+    monitor.record_slow_check_alert(100)
+    monitor.record_error_alert(ValueError("test error"))
 
-    # 过滤 warning 级别的告警
     warning_alerts = monitor.get_alerts(level="warning")
-    assert len(warning_alerts) > 0
-    assert all(a["level"] == "warning" for a in warning_alerts)
+    assert len(warning_alerts) == 1
+    assert all(alert["level"] == "warning" for alert in warning_alerts)
 
-    # 过滤 error 级别的告警
     error_alerts = monitor.get_alerts(level="error")
-    assert len(error_alerts) > 0
-    assert all(a["level"] == "error" for a in error_alerts)
+    assert len(error_alerts) == 1
+    assert all(alert["level"] == "error" for alert in error_alerts)
 
 
 @pytest.mark.asyncio
@@ -156,13 +101,10 @@ async def test_alert_limit():
     """测试告警数量限制"""
     monitor = PermissionMonitor()
 
-    # 生成大量告警
     for _ in range(50):
-        monitor.record_permission_check(allowed=True, elapsed_time=0.1)
+        monitor.record_slow_check_alert(100)
 
-    # 获取告警，限制为10条
-    alerts = monitor.get_alerts(limit=10)
-    assert len(alerts) == 10
+    assert len(monitor.get_alerts(limit=10)) == 10
 
 
 @pytest.mark.asyncio
@@ -170,17 +112,11 @@ async def test_metrics_reset():
     """测试指标重置"""
     monitor = PermissionMonitor()
 
-    # 记录一些指标
-    monitor.record_permission_check(allowed=True, elapsed_time=0.005)
     monitor.record_cache_hit()
-
-    assert monitor.metrics.check_metrics.total_checks == 1
     assert monitor.metrics.cache_metrics.total_accesses == 1
 
-    # 重置指标
     monitor.reset_metrics()
 
-    assert monitor.metrics.check_metrics.total_checks == 0
     assert monitor.metrics.cache_metrics.total_accesses == 0
 
 
@@ -189,15 +125,12 @@ async def test_alerts_clear():
     """测试告警清除"""
     monitor = PermissionMonitor()
 
-    # 生成告警
-    monitor.record_permission_check(allowed=True, elapsed_time=0.1)
+    monitor.record_slow_check_alert(100)
+    assert len(monitor.get_alerts()) == 1
 
-    assert len(monitor.get_alerts()) > 0
-
-    # 清除告警
     monitor.clear_alerts()
 
-    assert len(monitor.get_alerts()) == 0
+    assert monitor.get_alerts() == []
 
 
 @pytest.mark.asyncio
@@ -205,27 +138,16 @@ async def test_metrics_to_dict():
     """测试指标转换为字典"""
     monitor = PermissionMonitor()
 
-    # 记录指标
-    monitor.record_permission_check(allowed=True, elapsed_time=0.005)
-    monitor.record_permission_check(allowed=False, elapsed_time=0.003)
     monitor.record_cache_hit()
     monitor.record_cache_miss()
 
     metrics_dict = monitor.get_metrics()
 
-    assert "check_metrics" in metrics_dict
     assert "cache_metrics" in metrics_dict
     assert "uptime_seconds" in metrics_dict
-
-    check_metrics = metrics_dict["check_metrics"]
-    assert check_metrics["total_checks"] == 2
-    assert check_metrics["allowed_checks"] == 1
-    assert check_metrics["denied_checks"] == 1
-
-    cache_metrics = metrics_dict["cache_metrics"]
-    assert cache_metrics["total_accesses"] == 2
-    assert cache_metrics["cache_hits"] == 1
-    assert cache_metrics["cache_misses"] == 1
+    assert metrics_dict["cache_metrics"]["total_accesses"] == 2
+    assert metrics_dict["cache_metrics"]["cache_hits"] == 1
+    assert metrics_dict["cache_metrics"]["cache_misses"] == 1
 
 
 @pytest.mark.asyncio
@@ -234,7 +156,6 @@ async def test_global_monitor_instance():
     monitor1 = get_permission_monitor()
     monitor2 = get_permission_monitor()
 
-    # 应该是同一个实例
     assert monitor1 is monitor2
 
 
@@ -244,8 +165,7 @@ async def test_set_global_monitor():
     new_monitor = PermissionMonitor()
     set_permission_monitor(new_monitor)
 
-    monitor = get_permission_monitor()
-    assert monitor is new_monitor
+    assert get_permission_monitor() is new_monitor
 
 
 @pytest.mark.asyncio
@@ -254,87 +174,24 @@ async def test_alert_timestamp():
     monitor = PermissionMonitor()
 
     before_time = time.time()
-    monitor.record_permission_check(allowed=True, elapsed_time=0.1)
+    monitor.record_slow_check_alert(100)
     after_time = time.time()
 
     alerts = monitor.get_alerts()
-    assert len(alerts) > 0
     assert before_time <= alerts[0]["timestamp"] <= after_time
 
 
-@pytest.mark.asyncio
-async def test_alert_data():
-    """测试告警数据"""
-    monitor = PermissionMonitor()
-
-    # 记录慢速权限检查
-    monitor.record_permission_check(allowed=True, elapsed_time=0.1)
-
-    alerts = monitor.get_alerts()
-    assert len(alerts) > 0
-    assert "data" in alerts[0]
-    assert "elapsed_time" in alerts[0]["data"]
-
-
-@pytest.mark.asyncio
-async def test_permission_check_metrics_dataclass():
-    """测试权限检查指标数据类"""
-    metrics = PermissionCheckMetrics()
-
-    # 初始状态
-    assert metrics.total_checks == 0
-    assert metrics.avg_time == 0.0
-    assert metrics.allow_rate == 0.0
-
-    # 添加数据
-    metrics.total_checks = 100
-    metrics.allowed_checks = 80
-    metrics.denied_checks = 20
-    metrics.total_time = 0.5
-
-    assert metrics.avg_time == pytest.approx(0.005, 0.0001)
-    assert metrics.allow_rate == pytest.approx(80.0, 0.1)
-    assert metrics.deny_rate == pytest.approx(20.0, 0.1)
-
-
 @pytest.mark.asyncio
 async def test_cache_metrics_dataclass():
     """测试缓存指标数据类"""
     metrics = CacheMetrics()
 
-    # 初始状态
     assert metrics.total_accesses == 0
     assert metrics.hit_rate == 0.0
 
-    # 添加数据
     metrics.total_accesses = 100
     metrics.cache_hits = 80
     metrics.cache_misses = 20
 
     assert metrics.hit_rate == pytest.approx(80.0, 0.1)
     assert metrics.miss_rate == pytest.approx(20.0, 0.1)
-
-
-@pytest.mark.asyncio
-async def test_health_check_ignores_cache_hit_rate_without_samples():
-    """没有缓存访问样本时，不应判定缓存命中率过低"""
-    monitor = PermissionMonitor()
-    metrics = monitor.get_metrics()
-    cache_stats = monitor.get_cache_stats()
-
-    health = evaluate_permission_system_health(metrics, cache_stats)
-
-    assert metrics["cache_metrics"]["total_accesses"] == 0
-    assert "缓存命中率过低" not in health["issues"]
-
-
-@pytest.mark.asyncio
-async def test_health_check_ignores_cache_hit_rate_with_too_few_samples():
-    """缓存访问样本过少时，不应判定缓存命中率过低"""
-    monitor = PermissionMonitor()
-    for _ in range(3):
-        monitor.record_cache_miss()
-
-    health = evaluate_permission_system_health(monitor.get_metrics(), monitor.get_cache_stats())
-
-    assert "缓存命中率过低" not in health["issues"]

backend/tests/test_permission_monitoring_api.py

@@ -1,13 +1,11 @@
-"""监控API测试：权限系统监控API端点验证
+"""监控API测试：权限系统监控API端点验证。"""
 
-测试权限系统监控API的功能。
-"""
+import uuid
 
 import pytest
-from fastapi.testclient import TestClient
 from sqlalchemy import text
 
-from app.core.permission_monitor import get_permission_monitor, set_permission_monitor, PermissionMonitor
+from app.core.permission_monitor import set_permission_monitor, PermissionMonitor
 from app.api.v1 import permission_monitoring
 
 
@@ -20,202 +18,226 @@ class FakeIpInfo:
         self.location = f"中国 / {province} / {city} / 电信"
 
 
+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)})
+    ).scalar_one_or_none()
+    if not study_exists:
+        await db_session.execute(
+            text(
+                """
+                INSERT INTO studies (id, code, name, status, is_locked, visit_schedule, active_roles)
+                VALUES (:id, :code, :name, :status, :is_locked, :visit_schedule, :active_roles)
+                """
+            ),
+            {
+                "id": str(study_id),
+                "code": f"PERM-MON-{study_id.hex[:8]}",
+                "name": "Permission Monitoring Study",
+                "status": "ACTIVE",
+                "is_locked": False,
+                "visit_schedule": "[]",
+                "active_roles": "[]",
+            },
+        )
+
+    user_exists = (
+        await db_session.execute(text("SELECT id FROM users WHERE id = :id"), {"id": str(user_id)})
+    ).scalar_one_or_none()
+    if not user_exists:
+        await db_session.execute(
+            text(
+                """
+                INSERT INTO users (id, email, password_hash, full_name, role, clinical_department, status)
+                VALUES (:id, :email, :password_hash, :full_name, :role, :clinical_department, :status)
+                """
+            ),
+            {
+                "id": str(user_id),
+                "email": f"{user_id.hex}@example.com",
+                "password_hash": "hash",
+                "full_name": "Permission Monitoring User",
+                "role": "PM",
+                "clinical_department": "临床运营",
+                "status": "ACTIVE",
+            },
+        )
+
+    await db_session.execute(
+        text(
+            """
+            INSERT INTO permission_access_logs
+                (id, study_id, user_id, endpoint_key, role, allowed, elapsed_ms, ip_address, created_at)
+            VALUES
+                (:id, :study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
+            """
+        ),
+        {
+            "id": str(uuid.uuid4()),
+            "study_id": str(study_id),
+            "user_id": str(user_id),
+            "endpoint_key": "admin.permissions.read",
+            "role": "PM",
+            "allowed": allowed,
+            "elapsed_ms": elapsed_ms,
+            "ip_address": "127.0.0.1",
+        },
+    )
+    await db_session.commit()
+
+
 @pytest.mark.asyncio
-async def test_get_permission_metrics(client: TestClient, auth_headers: dict):
+async def test_get_permission_metrics(db_session):
     """测试获取权限系统指标"""
-    # 清除并重置监控器
     monitor = PermissionMonitor()
     set_permission_monitor(monitor)
 
-    # 记录一些指标
-    monitor.record_permission_check(allowed=True, elapsed_time=0.005)
-    monitor.record_permission_check(allowed=False, elapsed_time=0.003)
+    study_id = uuid.uuid4()
+    user_id = uuid.uuid4()
+    await _seed_permission_log(db_session, study_id, user_id, allowed=True, elapsed_ms=5)
+    await _seed_permission_log(db_session, study_id, user_id, allowed=False, elapsed_ms=3)
 
-    response = client.get("/api/v1/permission-monitoring/metrics", headers=auth_headers)
-    assert response.status_code == 200
-
-    data = response.json()
+    data = await permission_monitoring.get_permission_metrics(db=db_session, _=object(), hours=24)
     assert "check_metrics" in data
     assert "cache_metrics" in data
     assert data["check_metrics"]["total_checks"] == 2
+    assert data["check_metrics"]["allowed_checks"] == 1
+    assert data["check_metrics"]["denied_checks"] == 1
 
 
 @pytest.mark.asyncio
-async def test_get_cache_statistics(client: TestClient, auth_headers: dict):
+async def test_get_cache_statistics(db_session):
     """测试获取缓存统计"""
-    # 清除并重置监控器
     monitor = PermissionMonitor()
     set_permission_monitor(monitor)
 
-    # 记录缓存访问
     monitor.record_cache_hit()
     monitor.record_cache_hit()
     monitor.record_cache_miss()
 
-    response = client.get("/api/v1/permission-monitoring/cache-stats", headers=auth_headers)
-    assert response.status_code == 200
-
-    data = response.json()
+    data = await permission_monitoring.get_cache_statistics(_=object(), db=db_session)
     assert "cache_metrics" in data
     assert data["cache_metrics"]["total_accesses"] == 3
     assert data["cache_metrics"]["cache_hits"] == 2
 
 
 @pytest.mark.asyncio
-async def test_get_alerts(client: TestClient, auth_headers: dict):
+async def test_get_alerts(db_session):
     """测试获取告警列表"""
-    # 清除并重置监控器
     monitor = PermissionMonitor()
     set_permission_monitor(monitor)
 
-    # 生成告警
-    monitor.record_permission_check(allowed=True, elapsed_time=0.1)
+    monitor.record_slow_check_alert(100)
 
-    response = client.get("/api/v1/permission-monitoring/alerts", headers=auth_headers)
-    assert response.status_code == 200
-
-    data = response.json()
+    data = await permission_monitoring.get_alerts(_=object(), db=db_session)
     assert "alerts" in data
     assert data["total"] > 0
 
 
 @pytest.mark.asyncio
-async def test_get_alerts_with_level_filter(client: TestClient, auth_headers: dict):
+async def test_get_alerts_with_level_filter(db_session):
     """测试按级别过滤告警"""
-    # 清除并重置监控器
     monitor = PermissionMonitor()
     set_permission_monitor(monitor)
 
-    # 生成告警
-    monitor.record_permission_check(allowed=True, elapsed_time=0.1)
+    monitor.record_slow_check_alert(100)
+    monitor.record_error_alert(ValueError("test error"))
 
-    response = client.get(
-        "/api/v1/permission-monitoring/alerts?level=warning",
-        headers=auth_headers,
-    )
-    assert response.status_code == 200
-
-    data = response.json()
+    data = await permission_monitoring.get_alerts(level="warning", _=object(), db=db_session)
     assert "alerts" in data
+    assert all(alert["level"] == "warning" for alert in data["alerts"])
 
 
 @pytest.mark.asyncio
-async def test_get_alerts_with_limit(client: TestClient, auth_headers: dict):
+async def test_get_alerts_with_limit(db_session):
     """测试限制告警数量"""
-    # 清除并重置监控器
     monitor = PermissionMonitor()
     set_permission_monitor(monitor)
 
-    # 生成多个告警
     for _ in range(20):
-        monitor.record_permission_check(allowed=True, elapsed_time=0.1)
+        monitor.record_slow_check_alert(100)
 
-    response = client.get(
-        "/api/v1/permission-monitoring/alerts?limit=5",
-        headers=auth_headers,
-    )
-    assert response.status_code == 200
-
-    data = response.json()
+    data = await permission_monitoring.get_alerts(limit=5, _=object(), db=db_session)
     assert len(data["alerts"]) <= 5
 
 
 @pytest.mark.asyncio
-async def test_reset_metrics(client: TestClient, auth_headers: dict):
+async def test_reset_metrics(db_session):
     """测试重置指标"""
-    # 清除并重置监控器
     monitor = PermissionMonitor()
     set_permission_monitor(monitor)
 
-    # 记录指标
-    monitor.record_permission_check(allowed=True, elapsed_time=0.005)
-    assert monitor.metrics.check_metrics.total_checks == 1
+    monitor.record_cache_hit()
+    assert monitor.metrics.cache_metrics.total_accesses == 1
 
-    # 重置指标
-    response = client.post("/api/v1/permission-monitoring/reset-metrics", headers=auth_headers)
-    assert response.status_code == 200
+    result = await permission_monitoring.reset_metrics(_=object(), db=db_session)
 
-    # 验证指标已重置
-    assert monitor.metrics.check_metrics.total_checks == 0
+    assert result["message"] == "指标已重置"
+    assert monitor.metrics.cache_metrics.total_accesses == 0
 
 
 @pytest.mark.asyncio
-async def test_clear_alerts(client: TestClient, auth_headers: dict):
+async def test_clear_alerts(db_session):
     """测试清除告警"""
-    # 清除并重置监控器
     monitor = PermissionMonitor()
     set_permission_monitor(monitor)
 
-    # 生成告警
-    monitor.record_permission_check(allowed=True, elapsed_time=0.1)
+    monitor.record_slow_check_alert(100)
     assert len(monitor.get_alerts()) > 0
 
-    # 清除告警
-    response = client.post("/api/v1/permission-monitoring/clear-alerts", headers=auth_headers)
-    assert response.status_code == 200
+    result = await permission_monitoring.clear_alerts(_=object(), db=db_session)
 
-    # 验证告警已清除
+    assert result["message"] == "告警已清除"
     assert len(monitor.get_alerts()) == 0
 
 
 @pytest.mark.asyncio
-async def test_permission_system_health_healthy(client: TestClient, auth_headers: dict):
+async def test_permission_system_health_healthy(db_session):
     """测试权限系统健康检查（健康状态）"""
-    # 清除并重置监控器
     monitor = PermissionMonitor()
     set_permission_monitor(monitor)
 
-    # 记录良好的指标
-    for _ in range(100):
-        monitor.record_permission_check(allowed=True, elapsed_time=0.001)
     for _ in range(100):
         monitor.record_cache_hit()
 
-    response = client.get("/api/v1/permission-monitoring/health", headers=auth_headers)
-    assert response.status_code == 200
-
-    data = response.json()
-    assert data["status"] in ["healthy", "degraded"]
-    assert data["health_score"] > 50
+    data = await permission_monitoring.permission_system_health(db=db_session, _=object())
+    assert data["status"] == "healthy"
+    assert data["health_score"] >= 80
 
 
 @pytest.mark.asyncio
-async def test_permission_system_health_degraded(client: TestClient, auth_headers: dict):
+async def test_permission_system_health_degraded(db_session):
     """测试权限系统健康检查（降级状态）"""
-    # 清除并重置监控器
     monitor = PermissionMonitor()
     set_permission_monitor(monitor)
 
-    # 记录不良的指标
-    for _ in range(100):
-        monitor.record_permission_check(allowed=False, elapsed_time=0.1)
+    study_id = uuid.uuid4()
+    user_id = uuid.uuid4()
+    for _ in range(10):
+        await _seed_permission_log(db_session, study_id, user_id, allowed=False, elapsed_ms=100)
     for _ in range(100):
         monitor.record_cache_miss()
 
-    response = client.get("/api/v1/permission-monitoring/health", headers=auth_headers)
-    assert response.status_code == 200
-
-    data = response.json()
+    data = await permission_monitoring.permission_system_health(db=db_session, _=object())
     assert "status" in data
     assert "health_score" in data
     assert "issues" in data
+    assert "权限检查响应时间过长" in data["issues"]
+    assert "权限拒绝率过高" in data["issues"]
+    assert "缓存命中率过低" in data["issues"]
 
 
 @pytest.mark.asyncio
-async def test_permission_system_health_includes_metrics(client: TestClient, auth_headers: dict):
+async def test_permission_system_health_includes_metrics(db_session):
     """测试健康检查包含详细指标"""
-    # 清除并重置监控器
     monitor = PermissionMonitor()
     set_permission_monitor(monitor)
 
-    response = client.get("/api/v1/permission-monitoring/health", headers=auth_headers)
-    assert response.status_code == 200
-
-    data = response.json()
-    assert "metrics" in data
+    data = await permission_monitoring.permission_system_health(db=db_session, _=object())
+    assert "last_hour" in data
     assert "cache_stats" in data
-    assert "check_metrics" in data["metrics"]
-    assert "cache_metrics" in data["metrics"]
+    assert "total_checks" in data["last_hour"]
+    assert "cache_metrics" in data["cache_stats"]
 
 
 @pytest.mark.asyncio

backend/tests/test_prerequisite_permissions.py

@@ -70,11 +70,11 @@ async def test_prerequisite_permission_missing(db_session: AsyncSession):
 
 
 @pytest.mark.asyncio
-async def test_prerequisite_permission_not_configured(db_session: AsyncSession):
-    """测试前置权限未配置的情况"""
+async def test_default_prerequisite_permission_satisfies_when_not_overridden(db_session: AsyncSession):
+    """预设角色未配置前置权限时，应使用默认权限矩阵判断"""
     study_id = uuid.uuid4()
 
-    # 创建权限：主权限允许，前置权限未配置
+    # CRA 默认拥有 sites:read，因此未显式配置前置权限时仍满足前置条件。
     main_perm = ApiEndpointPermission(
         study_id=study_id,
         role="CRA",
@@ -88,6 +88,26 @@ async def test_prerequisite_permission_not_configured(db_session: AsyncSession):
     result = await role_has_api_permission(
         db_session, study_id, "CRA", "subjects:create", check_prerequisites=True
     )
+    assert result is True
+
+
+@pytest.mark.asyncio
+async def test_custom_role_prerequisite_permission_not_configured(db_session: AsyncSession):
+    """自定义角色无默认前置权限时，应被前置权限拦截"""
+    study_id = uuid.uuid4()
+
+    main_perm = ApiEndpointPermission(
+        study_id=study_id,
+        role="DATA_MANAGER",
+        endpoint_key="subjects:create",
+        allowed=True,
+    )
+    db_session.add(main_perm)
+    await db_session.commit()
+
+    result = await role_has_api_permission(
+        db_session, study_id, "DATA_MANAGER", "subjects:create", check_prerequisites=True
+    )
     assert result is False
 
 
@@ -323,8 +343,8 @@ async def test_prerequisite_with_no_prerequisites_operation(db_session: AsyncSes
 
 
 @pytest.mark.asyncio
-async def test_prerequisite_missing_not_configured(db_session: AsyncSession):
-    """测试前置权限未配置时的缺失检查"""
+async def test_default_prerequisite_not_reported_missing_when_not_overridden(db_session: AsyncSession):
+    """预设角色默认拥有前置权限时，不应报告缺失"""
     study_id = uuid.uuid4()
 
     # 创建权限：主权限允许，前置权限未配置
@@ -341,4 +361,24 @@ async def test_prerequisite_missing_not_configured(db_session: AsyncSession):
     missing = await get_missing_prerequisites(
         db_session, study_id, "CRA", "subjects:create"
     )
+    assert missing == []
+
+
+@pytest.mark.asyncio
+async def test_custom_role_prerequisite_missing_not_configured(db_session: AsyncSession):
+    """自定义角色无默认前置权限时，应报告缺失"""
+    study_id = uuid.uuid4()
+
+    main_perm = ApiEndpointPermission(
+        study_id=study_id,
+        role="DATA_MANAGER",
+        endpoint_key="subjects:create",
+        allowed=True,
+    )
+    db_session.add(main_perm)
+    await db_session.commit()
+
+    missing = await get_missing_prerequisites(
+        db_session, study_id, "DATA_MANAGER", "subjects:create"
+    )
     assert "sites:read" in missing

backend/tests/test_study_role_context.py

@@ -1,4 +1,5 @@
 import uuid
+from datetime import datetime, timezone
 
 import pytest
 
@@ -28,6 +29,8 @@ def _make_study() -> Study:
         status="ACTIVE",
         is_locked=False,
         visit_schedule=[],
+        active_roles=[],
+        created_at=datetime.now(timezone.utc),
     )