Cheng Zhou
453 lines
13 KiB
Markdown
453 lines
13 KiB
Markdown
# 权限系统测试指南
|
|
|
|
## 测试环境准备
|
|
|
|
### 1. 数据库迁移
|
|
```bash
|
|
cd backend
|
|
alembic upgrade head
|
|
```
|
|
|
|
### 2. 启动应用
|
|
```bash
|
|
uvicorn app.main:app --reload
|
|
```
|
|
|
|
### 3. 验证API端点注册
|
|
访问 `http://localhost:8000/api/v1/api-permissions/endpoints` 查看所有已注册的API端点。
|
|
|
|
## 单元测试
|
|
|
|
### 1. 权限检查函数测试
|
|
|
|
**文件**: `backend/tests/test_api_permissions.py`
|
|
|
|
```python
|
|
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`
|
|
|
|
```python
|
|
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`
|
|
|
|
```python
|
|
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`
|
|
|
|
```python
|
|
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. 权限变更流程
|
|
|
|
**场景**: 管理员修改权限后，用户权限立即生效
|
|
|
|
```python
|
|
@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创建不良事件时需要读取参与者信息
|
|
|
|
```python
|
|
@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. 权限检查性能
|
|
|
|
```python
|
|
@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. 模块级权限回退
|
|
|
|
```python
|
|
@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)
|
|
```
|
|
|
|
## 运行测试
|
|
|
|
```bash
|
|
# 运行所有测试
|
|
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%
|
|
- [ ] 没有安全漏洞
|
|
- [ ] 文档完整
|