Files
ctms/docs/guides/permission-system-testing.md
T

13 KiB

权限系统测试指南

测试环境准备

1. 数据库迁移

cd backend
alembic upgrade head

2. 启动应用

uvicorn app.main:app --reload

3. 验证API端点注册

访问 http://localhost:8000/api/v1/api-permissions/endpoints 查看所有已注册的API端点。

单元测试

1. 权限检查函数测试

文件: backend/tests/test_api_permissions.py

import pytest
from app.core.project_permissions import role_has_api_permission
from app.models.api_endpoint_permission import ApiEndpointPermission

@pytest.mark.asyncio
async def test_api_permission_check_allowed(db_session, study_id):
    """测试接口级权限检查 - 允许"""
    # 创建权限记录
    perm = ApiEndpointPermission(
        study_id=study_id,
        role="CRA",
        endpoint_key="POST:/subjects",
        allowed=True,
    )
    db_session.add(perm)
    await db_session.commit()
    
    # 验证权限
    result = await role_has_api_permission(
        db_session, study_id, "CRA", "POST:/subjects"
    )
    assert result is True

@pytest.mark.asyncio
async def test_api_permission_check_denied(db_session, study_id):
    """测试接口级权限检查 - 拒绝"""
    # 创建权限记录
    perm = ApiEndpointPermission(
        study_id=study_id,
        role="PV",
        endpoint_key="POST:/subjects",
        allowed=False,
    )
    db_session.add(perm)
    await db_session.commit()
    
    # 验证权限
    result = await role_has_api_permission(
        db_session, study_id, "PV", "POST:/subjects"
    )
    assert result is False

@pytest.mark.asyncio
async def test_api_permission_fallback_to_module_level(db_session, study_id):
    """测试权限回退 - 接口级权限未配置时回退到模块级权限"""
    # 不创建接口级权限,应该回退到模块级权限
    result = await role_has_api_permission(
        db_session, study_id, "CRA", "POST:/subjects"
    )
    # 结果取决于模块级权限配置
    assert isinstance(result, bool)

@pytest.mark.asyncio
async def test_admin_always_allowed(db_session, study_id):
    """测试ADMIN角色总是被允许"""
    result = await role_has_api_permission(
        db_session, study_id, "ADMIN", "POST:/subjects"
    )
    assert result is True

2. 权限配置测试

文件: backend/tests/test_api_permissions_config.py

import pytest
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, MODULE_TO_ENDPOINTS

def test_api_endpoint_permissions_structure():
    """测试API端点权限配置结构"""
    for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
        assert "module" in config
        assert "action" in config
        assert "description" in config
        assert "default_roles" in config
        assert config["action"] in ["read", "write"]
        assert isinstance(config["default_roles"], list)

def test_module_to_endpoints_mapping():
    """测试模块到端点的映射"""
    for module, actions in MODULE_TO_ENDPOINTS.items():
        assert "read" in actions
        assert "write" in actions
        assert isinstance(actions["read"], list)
        assert isinstance(actions["write"], list)
        
        # 验证所有端点都在API_ENDPOINT_PERMISSIONS中定义
        for endpoint_key in actions["read"] + actions["write"]:
            assert endpoint_key in API_ENDPOINT_PERMISSIONS

def test_subjects_endpoints_configured():
    """测试subjects模块的端点配置"""
    expected_endpoints = [
        "POST:/subjects",
        "GET:/subjects",
        "GET:/subjects/{id}",
        "PATCH:/subjects/{id}",
        "DELETE:/subjects/{id}",
    ]
    for endpoint_key in expected_endpoints:
        assert endpoint_key in API_ENDPOINT_PERMISSIONS
        assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "subjects"

def test_fees_endpoints_configured():
    """测试fees模块的端点配置"""
    expected_endpoints = [
        "POST:/fees/contracts",
        "GET:/fees/contracts",
        "GET:/fees/contracts/{id}",
        "PATCH:/fees/contracts/{id}",
        "DELETE:/fees/contracts/{id}",
        "POST:/fees/contracts/{id}/payments",
        "PATCH:/fees/payments/{id}",
        "DELETE:/fees/payments/{id}",
        "POST:/finance/contracts",
        "GET:/finance/contracts",
        "GET:/finance/contracts/{id}",
        "PATCH:/finance/contracts/{id}",
        "DELETE:/finance/contracts/{id}",
    ]
    for endpoint_key in expected_endpoints:
        assert endpoint_key in API_ENDPOINT_PERMISSIONS
        assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "fees"

集成测试

1. 权限管理API测试

文件: backend/tests/test_api_permissions_endpoints.py

import pytest
from fastapi.testclient import TestClient

@pytest.mark.asyncio
async def test_list_api_endpoints(client: TestClient, admin_token: str):
    """测试获取所有API端点"""
    response = client.get(
        "/api/v1/api-permissions/endpoints",
        headers={"Authorization": f"Bearer {admin_token}"}
    )
    assert response.status_code == 200
    data = response.json()
    assert "endpoints" in data
    assert len(data["endpoints"]) > 0
    
    # 验证端点结构
    for endpoint in data["endpoints"]:
        assert "endpoint_key" in endpoint
        assert "method" in endpoint
        assert "path" in endpoint
        assert "module" in endpoint
        assert "action" in endpoint

@pytest.mark.asyncio
async def test_get_study_api_permissions(client: TestClient, pm_token: str, study_id: str):
    """测试获取项目的权限矩阵"""
    response = client.get(
        f"/api/v1/studies/{study_id}/api-permissions",
        headers={"Authorization": f"Bearer {pm_token}"}
    )
    assert response.status_code == 200
    data = response.json()
    
    # 验证权限矩阵结构
    for role, endpoints in data.items():
        assert isinstance(endpoints, dict)
        for endpoint_key, permission in endpoints.items():
            assert "allowed" in permission
            assert isinstance(permission["allowed"], bool)

@pytest.mark.asyncio
async def test_update_study_api_permissions(client: TestClient, pm_token: str, study_id: str):
    """测试更新项目的权限矩阵"""
    payload = {
        "CRA": {
            "POST:/subjects": True,
            "GET:/subjects": True,
            "PATCH:/subjects/{id}": True,
        },
        "PV": {
            "GET:/subjects": True,
            "GET:/subjects/{id}": True,
        }
    }
    
    response = client.put(
        f"/api/v1/studies/{study_id}/api-permissions",
        json=payload,
        headers={"Authorization": f"Bearer {pm_token}"}
    )
    assert response.status_code == 200
    
    # 验证权限已更新
    response = client.get(
        f"/api/v1/studies/{study_id}/api-permissions",
        headers={"Authorization": f"Bearer {pm_token}"}
    )
    data = response.json()
    assert data["CRA"]["POST:/subjects"]["allowed"] is True
    assert data["PV"]["POST:/subjects"]["allowed"] is False

2. 迁移模块功能测试

文件: backend/tests/test_migrated_endpoints.py

import pytest
from fastapi.testclient import TestClient

@pytest.mark.asyncio
async def test_create_subject_with_api_permission(client: TestClient, cra_token: str, study_id: str):
    """测试创建参与者 - 使用接口级权限"""
    payload = {
        "subject_no": "SUBJ001",
        "site_id": "site-uuid",
        "status": "ACTIVE",
    }
    
    response = client.post(
        f"/api/v1/studies/{study_id}/subjects",
        json=payload,
        headers={"Authorization": f"Bearer {cra_token}"}
    )
    assert response.status_code == 201

@pytest.mark.asyncio
async def test_create_subject_without_permission(client: TestClient, pv_token: str, study_id: str):
    """测试创建参与者 - 权限不足"""
    payload = {
        "subject_no": "SUBJ001",
        "site_id": "site-uuid",
        "status": "ACTIVE",
    }
    
    response = client.post(
        f"/api/v1/studies/{study_id}/subjects",
        json=payload,
        headers={"Authorization": f"Bearer {pv_token}"}
    )
    assert response.status_code == 403

@pytest.mark.asyncio
async def test_list_subjects_with_permission(client: TestClient, cra_token: str, study_id: str):
    """测试查询参与者列表 - 有权限"""
    response = client.get(
        f"/api/v1/studies/{study_id}/subjects",
        headers={"Authorization": f"Bearer {cra_token}"}
    )
    assert response.status_code == 200

@pytest.mark.asyncio
async def test_create_ae_with_permission(client: TestClient, cra_token: str, study_id: str):
    """测试创建不良事件 - 使用接口级权限"""
    payload = {
        "term": "Headache",
        "onset_date": "2026-05-13",
        "seriousness": "MILD",
    }
    
    response = client.post(
        f"/api/v1/studies/{study_id}/aes",
        json=payload,
        headers={"Authorization": f"Bearer {cra_token}"}
    )
    assert response.status_code == 201

@pytest.mark.asyncio
async def test_create_contract_fee_with_permission(client: TestClient, pm_token: str):
    """测试创建费用合同 - 使用接口级权限"""
    payload = {
        "projectId": "project-uuid",
        "centerId": "center-uuid",
        "contractAmount": 10000,
        "contractCases": 100,
    }
    
    response = client.post(
        "/api/v1/fees/contracts",
        json=payload,
        headers={"Authorization": f"Bearer {pm_token}"}
    )
    assert response.status_code == 201

端到端测试

1. 权限变更流程

场景: 管理员修改权限后,用户权限立即生效

@pytest.mark.asyncio
async def test_permission_change_takes_effect_immediately(
    client: TestClient, pm_token: str, cra_token: str, study_id: str
):
    """测试权限变更立即生效"""
    # 1. 初始状态:CRA可以创建参与者
    payload = {"subject_no": "SUBJ001", "site_id": "site-uuid"}
    response = client.post(
        f"/api/v1/studies/{study_id}/subjects",
        json=payload,
        headers={"Authorization": f"Bearer {cra_token}"}
    )
    assert response.status_code == 201
    
    # 2. PM修改权限:禁止CRA创建参与者
    perm_payload = {
        "CRA": {
            "POST:/subjects": False,
        }
    }
    response = client.put(
        f"/api/v1/studies/{study_id}/api-permissions",
        json=perm_payload,
        headers={"Authorization": f"Bearer {pm_token}"}
    )
    assert response.status_code == 200
    
    # 3. CRA尝试创建参与者:应该被拒绝
    payload = {"subject_no": "SUBJ002", "site_id": "site-uuid"}
    response = client.post(
        f"/api/v1/studies/{study_id}/subjects",
        json=payload,
        headers={"Authorization": f"Bearer {cra_token}"}
    )
    assert response.status_code == 403

2. 跨模块数据访问

场景: PV创建不良事件时需要读取参与者信息

@pytest.mark.asyncio
async def test_cross_module_data_access(
    client: TestClient, pv_token: str, study_id: str, subject_id: str
):
    """测试跨模块数据访问权限"""
    # 1. PV查询参与者信息(需要GET:/subjects/{id}权限)
    response = client.get(
        f"/api/v1/studies/{study_id}/subjects/{subject_id}",
        headers={"Authorization": f"Bearer {pv_token}"}
    )
    assert response.status_code == 200
    
    # 2. PV创建不良事件(需要POST:/risk-issues权限)
    payload = {
        "term": "Headache",
        "onset_date": "2026-05-13",
        "seriousness": "MILD",
        "subject_id": subject_id,
    }
    response = client.post(
        f"/api/v1/studies/{study_id}/aes",
        json=payload,
        headers={"Authorization": f"Bearer {pv_token}"}
    )
    assert response.status_code == 201

性能测试

1. 权限检查性能

@pytest.mark.asyncio
async def test_permission_check_performance(db_session, study_id):
    """测试权限检查性能"""
    import time
    
    start = time.time()
    for _ in range(1000):
        await role_has_api_permission(
            db_session, study_id, "CRA", "POST:/subjects"
        )
    elapsed = time.time() - start
    
    # 1000次权限检查应该在1秒内完成
    assert elapsed < 1.0

向后兼容测试

1. 模块级权限回退

@pytest.mark.asyncio
async def test_fallback_to_module_level_permission(db_session, study_id):
    """测试回退到模块级权限"""
    # 不创建接口级权限,应该使用模块级权限
    result = await role_has_api_permission(
        db_session, study_id, "CRA", "POST:/subjects"
    )
    # 结果应该基于模块级权限配置
    assert isinstance(result, bool)

运行测试

# 运行所有测试
pytest backend/tests/

# 运行特定测试文件
pytest backend/tests/test_api_permissions.py

# 运行特定测试
pytest backend/tests/test_api_permissions.py::test_api_permission_check_allowed

# 运行并显示覆盖率
pytest backend/tests/ --cov=app --cov-report=html

验证清单

  • 所有单元测试通过
  • 所有集成测试通过
  • 所有端到端测试通过
  • 性能测试通过(权限检查 < 1ms)
  • 向后兼容测试通过
  • 代码覆盖率 > 80%
  • 没有安全漏洞
  • 文档完整