整理测试代码并清理文档引用
This commit is contained in:
Cheng Zhou
@@ -1,11 +0,0 @@
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def test_new_ae_records_default_to_follow_up_status():
|
||||
crud_source = Path("app/crud/ae.py").read_text(encoding="utf-8")
|
||||
model_source = Path("app/models/ae.py").read_text(encoding="utf-8")
|
||||
|
||||
assert 'status="FOLLOW_UP"' in crud_source
|
||||
assert 'default="FOLLOW_UP"' in model_source
|
||||
assert 'status="NEW"' not in crud_source
|
||||
assert 'default="NEW"' not in model_source
|
||||
@@ -1,86 +0,0 @@
|
||||
"""合同费用基础字段测试"""
|
||||
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
import uuid
|
||||
|
||||
from starlette.requests import Request
|
||||
|
||||
from app.api.v1 import fees_contracts
|
||||
from app.models.contract_fee import ContractFee
|
||||
from app.schemas.contract_fee import ContractFeeCreate, ContractFeeRead, ContractFeeUpdate
|
||||
|
||||
|
||||
def _request_with_query(query: str) -> Request:
|
||||
return Request(
|
||||
{
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/api/v1/fees/contracts",
|
||||
"query_string": query.encode(),
|
||||
"headers": [],
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_contract_fee_list_query_accepts_legacy_project_params():
|
||||
"""合同费用列表应兼容旧 projectId/centerId 查询参数。"""
|
||||
study_id = uuid.uuid4()
|
||||
center_id = uuid.uuid4()
|
||||
resolver = getattr(fees_contracts, "_resolve_contract_fee_list_query", None)
|
||||
|
||||
assert resolver is not None
|
||||
resolved_study_id, resolved_center_id = resolver(
|
||||
_request_with_query(f"projectId={study_id}¢erId={center_id}"),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
|
||||
assert resolved_study_id == study_id
|
||||
assert resolved_center_id == center_id
|
||||
|
||||
|
||||
def test_contract_fee_schema_includes_contract_basic_fields():
|
||||
"""合同费用应包含合同基础信息字段。"""
|
||||
study_id = uuid.uuid4()
|
||||
center_id = uuid.uuid4()
|
||||
|
||||
payload = ContractFeeCreate(
|
||||
study_id=study_id,
|
||||
center_id=center_id,
|
||||
contract_no="CT-001",
|
||||
signed_date=date(2026, 5, 27),
|
||||
contract_amount=Decimal("120000.00"),
|
||||
currency="CNY",
|
||||
remark="首版合同",
|
||||
contract_cases=12,
|
||||
)
|
||||
|
||||
assert payload.contract_no == "CT-001"
|
||||
assert payload.signed_date == date(2026, 5, 27)
|
||||
assert payload.currency == "CNY"
|
||||
assert payload.remark == "首版合同"
|
||||
assert payload.study_id == study_id
|
||||
assert "project_id" not in ContractFeeCreate.model_fields
|
||||
|
||||
update_payload = ContractFeeUpdate(contract_no="CT-002", currency="USD", remark="变更合同信息")
|
||||
assert update_payload.model_dump(exclude_unset=True) == {
|
||||
"contract_no": "CT-002",
|
||||
"currency": "USD",
|
||||
"remark": "变更合同信息",
|
||||
}
|
||||
|
||||
for field in ("contract_no", "signed_date", "currency", "remark"):
|
||||
assert field in ContractFeeRead.model_fields
|
||||
assert "study_id" in ContractFeeRead.model_fields
|
||||
assert "project_id" not in ContractFeeRead.model_fields
|
||||
|
||||
|
||||
def test_contract_fee_model_includes_contract_basic_columns():
|
||||
"""合同费用模型应持久化合同基础信息。"""
|
||||
columns = ContractFee.__table__.columns
|
||||
|
||||
assert "contract_no" in columns
|
||||
assert "signed_date" in columns
|
||||
assert "currency" in columns
|
||||
assert "remark" in columns
|
||||
@@ -1,81 +0,0 @@
|
||||
import uuid
|
||||
from datetime import date
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.schemas.drug_shipment import DrugShipmentCreate
|
||||
|
||||
|
||||
def shipment_payload(**overrides):
|
||||
payload = {
|
||||
"direction": "SEND",
|
||||
"center_id": uuid.uuid4(),
|
||||
"ship_date": date(2026, 6, 4),
|
||||
"receive_date": None,
|
||||
"quantity": 1,
|
||||
"batch_no": "DP-001",
|
||||
"carrier": "顺丰",
|
||||
"tracking_no": "SF100003",
|
||||
"status": "IN_TRANSIT",
|
||||
"remark": None,
|
||||
}
|
||||
payload.update(overrides)
|
||||
return payload
|
||||
|
||||
|
||||
def test_create_allows_pending_shipment_execution_fields_to_be_empty():
|
||||
shipment = DrugShipmentCreate.model_validate(
|
||||
shipment_payload(
|
||||
status="PENDING",
|
||||
ship_date=None,
|
||||
quantity=None,
|
||||
batch_no=None,
|
||||
carrier=None,
|
||||
tracking_no=None,
|
||||
receive_date=None,
|
||||
remark=None,
|
||||
)
|
||||
)
|
||||
|
||||
assert shipment.ship_date is None
|
||||
assert shipment.quantity is None
|
||||
assert shipment.batch_no is None
|
||||
assert shipment.carrier is None
|
||||
assert shipment.tracking_no is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("field_name", ["ship_date", "quantity", "batch_no", "carrier", "tracking_no"])
|
||||
def test_create_requires_shipment_execution_fields_after_pending(field_name):
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
DrugShipmentCreate.model_validate(shipment_payload(status="IN_TRANSIT", **{field_name: None}))
|
||||
|
||||
assert "发运信息" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_create_allows_pending_receipt_fields_to_be_empty():
|
||||
shipment = DrugShipmentCreate.model_validate(shipment_payload())
|
||||
|
||||
assert shipment.receive_date is None
|
||||
assert shipment.remark is None
|
||||
|
||||
|
||||
def test_create_requires_receive_date_when_signed():
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
DrugShipmentCreate.model_validate(shipment_payload(status="SIGNED", receive_date=None))
|
||||
|
||||
assert "接收日期" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_create_rejects_removed_returned_status():
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
DrugShipmentCreate.model_validate(shipment_payload(status="RETURNED", receive_date=date(2026, 6, 5)))
|
||||
|
||||
assert "无效状态" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_create_requires_remark_when_exceptional():
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
DrugShipmentCreate.model_validate(shipment_payload(status="EXCEPTION", remark=""))
|
||||
|
||||
assert "备注" in str(exc_info.value)
|
||||
@@ -1,39 +0,0 @@
|
||||
import uuid
|
||||
|
||||
from sqlalchemy import text
|
||||
import pytest
|
||||
|
||||
from app.crud.faq_category import create_category
|
||||
from app.crud.faq_item import create_item
|
||||
from app.schemas.faq import CategoryCreate, FaqCreate
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_category_defaults_to_active(db_session):
|
||||
result = await db_session.execute(text("SELECT id FROM studies LIMIT 1"))
|
||||
study_id = result.scalar_one()
|
||||
|
||||
category = await create_category(
|
||||
db_session,
|
||||
CategoryCreate(study_id=study_id, name="AE", sort_order=0),
|
||||
)
|
||||
|
||||
assert category.is_active is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_item_defaults_to_active(db_session):
|
||||
result = await db_session.execute(text("SELECT id FROM studies LIMIT 1"))
|
||||
study_id = result.scalar_one()
|
||||
category = await create_category(
|
||||
db_session,
|
||||
CategoryCreate(study_id=study_id, name=f"AE-{uuid.uuid4().hex[:8]}", sort_order=0),
|
||||
)
|
||||
|
||||
item = await create_item(
|
||||
db_session,
|
||||
FaqCreate(study_id=study_id, category_id=category.id, question="AE 如何记录？"),
|
||||
created_by=uuid.uuid4(),
|
||||
)
|
||||
|
||||
assert item.is_active is True
|
||||
@@ -1,37 +0,0 @@
|
||||
"""单元测试：IP 属地解析服务"""
|
||||
|
||||
from app.services.ip_location import Ip2RegionResolver
|
||||
|
||||
|
||||
class FakeSearcher:
|
||||
def search(self, _ip: str) -> str:
|
||||
return "中国|广东省|深圳市|电信|CN"
|
||||
|
||||
|
||||
def test_ip_location_handles_special_addresses():
|
||||
resolver = Ip2RegionResolver(db_path="/not-exists/ip2region.xdb")
|
||||
|
||||
assert resolver.lookup(None).location == "未知"
|
||||
assert resolver.lookup("not-an-ip").location == "未知"
|
||||
assert resolver.lookup("127.0.0.1").location == "本机"
|
||||
assert resolver.lookup("192.168.1.1").location == "局域网"
|
||||
|
||||
|
||||
def test_ip2region_result_parses_province_city_isp():
|
||||
resolver = Ip2RegionResolver(db_path="/not-exists/ip2region.xdb")
|
||||
resolver._searchers[4] = FakeSearcher()
|
||||
|
||||
result = resolver.lookup("8.8.8.8")
|
||||
|
||||
assert result.country == "中国"
|
||||
assert result.province == "广东省"
|
||||
assert result.city == "深圳市"
|
||||
assert result.isp == "电信"
|
||||
assert result.location == "中国 / 广东省 / 深圳市 / 电信"
|
||||
|
||||
|
||||
def test_default_resolver_can_use_packaged_xdb():
|
||||
resolver = Ip2RegionResolver()
|
||||
result = resolver.lookup("8.8.8.8")
|
||||
|
||||
assert result.location not in {"公网", "未知"}
|
||||
@@ -1,272 +0,0 @@
|
||||
"""
|
||||
第2批模块迁移测试：members 和 sites 模块
|
||||
测试接口级权限系统在 members 和 sites 模块中的应用
|
||||
"""
|
||||
|
||||
import uuid
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.project_permissions import role_has_api_permission
|
||||
from app.models.api_endpoint_permission import ApiEndpointPermission
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_member_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以添加项目成员"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="project_members:create",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "project_members:create")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_member_without_permission(db_session: AsyncSession):
|
||||
"""验证无权限的CRA无法添加项目成员"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限（拒绝）
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="project_members:create",
|
||||
allowed=False,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "project_members:create")
|
||||
assert allowed is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_members_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以查询项目成员列表"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="project_members:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "project_members:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_member_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以更新项目成员"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="project_members:update",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "project_members:update")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_member_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以删除项目成员"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="project_members:delete",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "project_members:delete")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_member_candidates_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以查询项目成员候选人"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="project_members:candidates",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "project_members:candidates")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_site_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以创建中心"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="sites:create",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "sites:create")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_sites_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以查询中心列表"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="sites:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "sites:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_sites_cra_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以查询中心列表"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="sites:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "sites:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_site_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以更新中心"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="sites:update",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "sites:update")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_site_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以删除中心"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="sites:delete",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "sites:delete")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_members_permission_denied_for_cra(db_session: AsyncSession):
|
||||
"""验证CRA无法执行成员管理操作"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限（拒绝）
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="project_members:create",
|
||||
allowed=False,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "project_members:create")
|
||||
assert allowed is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sites_permission_denied_for_cra_write(db_session: AsyncSession):
|
||||
"""验证CRA无法执行中心写操作"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
# 创建权限（拒绝）
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="sites:create",
|
||||
allowed=False,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
# 验证权限检查
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "sites:create")
|
||||
assert allowed is False
|
||||
|
||||
@@ -1,489 +0,0 @@
|
||||
"""
|
||||
第3批模块迁移测试：12个模块，63个端点
|
||||
测试接口级权限系统在所有第3批模块中的应用
|
||||
"""
|
||||
|
||||
import uuid
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.project_permissions import role_has_api_permission
|
||||
from app.models.api_endpoint_permission import ApiEndpointPermission
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 启动管理 (startup)
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_ethics_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以创建伦理记录"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="startup_ethics:create",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "startup_ethics:create")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_ethics_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以查询伦理记录列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="startup_ethics:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "startup_ethics:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_initiation_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以创建立项记录"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="startup_initiation:create",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "startup_initiation:create")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_auth_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以创建启动会或培训授权"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="startup_auth:create",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "startup_auth:create")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_auth_read_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以查询启动会或培训授权"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="startup_auth:read",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "startup_auth:read")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 项目权限管理已迁移为系统级权限
|
||||
# ============================================================================
|
||||
|
||||
def test_project_permissions_are_not_project_matrix_permissions():
|
||||
"""项目权限配置由 system:permissions:project_config 控制，不再进入项目矩阵。"""
|
||||
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, SYSTEM_PERMISSIONS
|
||||
|
||||
assert "permissions:read" not in API_ENDPOINT_PERMISSIONS
|
||||
assert "permissions:update" not in API_ENDPOINT_PERMISSIONS
|
||||
assert "system:permissions:project_config" in SYSTEM_PERMISSIONS
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 项目概览 (overview) - 1个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_overview_get_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以查询项目概览"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="project_overview:read",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "project_overview:read")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 监查访视问题汇总 (monitoring_visit_issues)
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_monitoring_issues_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以查询监查访视问题列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="monitoring_issues:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "monitoring_issues:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 药物发货 (drug_shipments) - 5个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_drug_shipments_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CTA可以创建药物发货"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CTA",
|
||||
endpoint_key="drug_shipments:create",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CTA", "drug_shipments:create")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_drug_shipments_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CTA可以查询药物发货列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CTA",
|
||||
endpoint_key="drug_shipments:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CTA", "drug_shipments:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 设备管理 (material_equipments) - 5个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_material_equipments_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以创建设备"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="material_equipments:create",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "material_equipments:create")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_material_equipments_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CTA可以查询设备列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CTA",
|
||||
endpoint_key="material_equipments:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CTA", "material_equipments:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 参与者PD (subject_pds) - 4个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subject_pds_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以创建参与者PD"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="subject_pds:create",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "subject_pds:create")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subject_pds_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以查询参与者PD列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="subject_pds:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "subject_pds:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 审计日志 (audit_logs) - 3个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_audit_logs_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以查询审计日志列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="audit_logs:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "audit_logs:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_audit_logs_export_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以导出审计日志"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="audit_logs:export",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "audit_logs:export")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 访视管理 (visits) - 5个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_visits_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PV可以创建访视"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PV",
|
||||
endpoint_key="visits:create",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PV", "visits:create")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_visits_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PV可以查询访视列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PV",
|
||||
endpoint_key="visits:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PV", "visits:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 注意事项 (precautions) - 5个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_precautions_create_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的QA可以创建注意事项"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="QA",
|
||||
endpoint_key="precautions:create",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "QA", "precautions:create")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_precautions_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的QA可以查询注意事项列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="QA",
|
||||
endpoint_key="precautions:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "QA", "precautions:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 病史记录 (subject_histories)
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subject_histories_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以查询病史记录列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="subject_histories:list",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "subject_histories:list")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_subject_histories_read_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的CRA可以查询病史记录详情"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="subject_histories:read",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "subject_histories:read")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 项目里程碑 (project_milestones) - 2个端点
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_milestones_list_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以查询项目里程碑列表"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="project_milestones:read",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "project_milestones:read")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_milestones_update_with_permission(db_session: AsyncSession):
|
||||
"""验证有权限的PM可以更新项目里程碑"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="PM",
|
||||
endpoint_key="project_milestones:update",
|
||||
allowed=True,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "PM", "project_milestones:update")
|
||||
assert allowed is True
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 权限拒绝场景
|
||||
# ============================================================================
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_startup_permission_denied_for_cra(db_session: AsyncSession):
|
||||
"""验证CRA无法执行启动管理操作"""
|
||||
study_id = uuid.uuid4()
|
||||
perm = ApiEndpointPermission(
|
||||
study_id=study_id,
|
||||
role="CRA",
|
||||
endpoint_key="startup_ethics:create",
|
||||
allowed=False,
|
||||
)
|
||||
db_session.add(perm)
|
||||
await db_session.commit()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "startup_ethics:create")
|
||||
assert allowed is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_removed_project_permissions_are_denied_for_project_roles(db_session: AsyncSession):
|
||||
"""项目角色不能再通过 permissions:update 这种残留 key 获得权限管理能力。"""
|
||||
study_id = uuid.uuid4()
|
||||
|
||||
allowed = await role_has_api_permission(db_session, study_id, "CRA", "permissions:update")
|
||||
assert allowed is False
|
||||
@@ -1,107 +0,0 @@
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def test_is_locked_migration_is_idempotent():
|
||||
source = Path("alembic/versions/20260116_01_add_is_locked_to_studies.py").read_text(encoding="utf-8")
|
||||
|
||||
assert 'if "is_locked" not in columns:' in source
|
||||
assert 'if "is_locked" in columns:' in source
|
||||
|
||||
|
||||
def test_removed_workflow_tables_are_dropped_conditionally():
|
||||
source = Path("alembic/versions/20260116_05_remove_document_workflows.py").read_text(encoding="utf-8")
|
||||
|
||||
assert 'if "workflow_actions" in tables:' in source
|
||||
assert 'if "version_workflows" in tables:' in source
|
||||
assert 'if "workflow_nodes" in tables:' in source
|
||||
assert 'if "workflow_templates" in tables:' in source
|
||||
|
||||
|
||||
def test_migration_state_check_script_exists():
|
||||
source = Path("scripts/check_migration_state.py").read_text(encoding="utf-8")
|
||||
|
||||
assert "missing alembic_version table" in source
|
||||
assert "studies" in source
|
||||
assert "subjects" in source
|
||||
assert "monitoring_visit_issues" in source
|
||||
|
||||
|
||||
def test_remove_qa_role_migration_casts_json_permissions_for_key_lookup():
|
||||
source = Path("alembic/versions/20260521_01_remove_qa_role.py").read_text(encoding="utf-8")
|
||||
|
||||
assert "permissions ? 'QA'" not in source
|
||||
assert "permissions::jsonb ? 'QA'" in source
|
||||
|
||||
|
||||
def test_role_template_copy_migration_updates_display_copy_for_current_role_keys():
|
||||
source = Path("alembic/versions/20260522_01_update_role_template_copy.py").read_text(encoding="utf-8")
|
||||
|
||||
assert "项目负责人，统筹项目全局，协调进度、资源与关键决策。" in source
|
||||
assert "负责各中心临床监查执行，跟进现场质量、数据和问题闭环。" in source
|
||||
assert "负责合同、药品及相关项目事务管理，保障执行支持与物资协同。" in source
|
||||
assert "负责医学审核与稽查，关注质量风险、合规性和医学一致性。" in source
|
||||
assert "负责药物警戒相关工作，跟踪安全性事件并支持风险评估。" in source
|
||||
assert '"IMP": ("CTA"' in source
|
||||
assert '"MEDICAL_REVIEW": ("QA"' in source
|
||||
assert "category = 'QA'" not in source
|
||||
|
||||
|
||||
def test_permission_template_migrations_do_not_seed_stale_startup_permissions():
|
||||
stale_keys = (
|
||||
"budget:create",
|
||||
"budget:list",
|
||||
"budget:read",
|
||||
"budget:update",
|
||||
"budget:delete",
|
||||
"timeline:create",
|
||||
"timeline:list",
|
||||
"timeline:read",
|
||||
"timeline:update",
|
||||
"timeline:delete",
|
||||
)
|
||||
for path in Path("alembic/versions").glob("*.py"):
|
||||
if path.name == "20260527_04_remove_stale_startup_permissions.py":
|
||||
continue
|
||||
source = path.read_text(encoding="utf-8")
|
||||
for key in stale_keys:
|
||||
assert key not in source, f"{key} should not be seeded by {path}"
|
||||
|
||||
|
||||
def test_legacy_startup_ethics_permission_keys_are_removed_by_followup_migration():
|
||||
source = Path("alembic/versions/20260527_05_remove_legacy_startup_ethics_permission_keys.py").read_text(encoding="utf-8")
|
||||
|
||||
assert '"feasibility:create"' in source
|
||||
assert '"feasibility:list"' in source
|
||||
assert '"feasibility:read"' in source
|
||||
assert '"feasibility:update"' in source
|
||||
assert '"feasibility:delete"' in source
|
||||
assert '"ethics:create"' in source
|
||||
assert '"ethics:list"' in source
|
||||
assert '"ethics:read"' in source
|
||||
assert '"ethics:update"' in source
|
||||
assert '"ethics:delete"' in source
|
||||
assert "startup_initiation:" not in source
|
||||
assert "startup_ethics:" not in source
|
||||
assert "DELETE FROM api_endpoint_permissions" in source
|
||||
assert "api_endpoint_permissions" in source
|
||||
assert "permission_templates" in source
|
||||
assert "permission_template_versions" in source
|
||||
|
||||
|
||||
def test_precautions_migration_renames_table_and_attachment_entity_type():
|
||||
source = Path("alembic/versions/20260527_08_rename_knowledge_notes_to_precautions.py").read_text(encoding="utf-8")
|
||||
|
||||
assert 'rename_table("knowledge_notes", "precautions")' in source
|
||||
assert "entity_type = 'precaution'" in source
|
||||
assert "entity_type = 'knowledge_note'" in source
|
||||
assert "api_endpoint_permissions" in source
|
||||
assert "permission_templates" in source
|
||||
|
||||
|
||||
def test_etmf_migration_reuses_existing_document_scope_type():
|
||||
source = Path("alembic/versions/20260527_09_add_etmf_nodes.py").read_text(encoding="utf-8")
|
||||
|
||||
assert "postgresql.ENUM(" in source
|
||||
assert 'name="document_scope_type"' in source
|
||||
assert "create_type=False" in source
|
||||
assert "scope_type = sa.Enum" not in source
|
||||
@@ -1,72 +0,0 @@
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||||
|
||||
from app.crud import monitoring_visit_issue as issue_crud
|
||||
from app.models.monitoring_visit_issue import MonitoringVisitIssue
|
||||
from app.schemas.monitoring_visit_issue import MonitoringVisitIssueCreate
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_monitoring_visit_issue_template_fields_can_be_filtered(tmp_path):
|
||||
db_path = tmp_path / "monitoring-issues.db"
|
||||
engine = create_async_engine(f"sqlite+aiosqlite:///{db_path}", future=True)
|
||||
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
||||
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(MonitoringVisitIssue.__table__.create)
|
||||
|
||||
study_id = uuid.UUID("11111111-1111-1111-1111-111111111111")
|
||||
site_id = uuid.UUID("22222222-2222-2222-2222-222222222222")
|
||||
async with SessionLocal() as session:
|
||||
created = await issue_crud.create_issue(
|
||||
session,
|
||||
study_id,
|
||||
MonitoringVisitIssueCreate(
|
||||
issue_no="MV-001",
|
||||
site_id=site_id,
|
||||
category="原始记录",
|
||||
subject_code="SUBJ-001",
|
||||
status="OPEN",
|
||||
severity="严重",
|
||||
mark="SDV",
|
||||
visit_cycle="V1",
|
||||
center_query="请补充签名日期",
|
||||
center_latest_reply="待中心回复",
|
||||
rectification_completed=False,
|
||||
due_at=datetime(2026, 5, 1, tzinfo=timezone.utc),
|
||||
),
|
||||
created_by=None,
|
||||
)
|
||||
|
||||
assert created.site_id == site_id
|
||||
assert created.severity == "严重"
|
||||
assert created.mark == "SDV"
|
||||
assert created.visit_cycle == "V1"
|
||||
assert created.center_query == "请补充签名日期"
|
||||
assert created.center_latest_reply == "待中心回复"
|
||||
assert created.rectification_completed is False
|
||||
|
||||
matched = await issue_crud.list_issues(
|
||||
session,
|
||||
study_id,
|
||||
site_id=site_id,
|
||||
severity="严重",
|
||||
mark="SDV",
|
||||
visit_cycle="V1",
|
||||
rectification_completed=False,
|
||||
due_from=datetime(2026, 5, 1, tzinfo=timezone.utc).date(),
|
||||
due_to=datetime(2026, 5, 1, tzinfo=timezone.utc).date(),
|
||||
)
|
||||
unmatched = await issue_crud.list_issues(
|
||||
session,
|
||||
study_id,
|
||||
site_id=uuid.UUID("33333333-3333-3333-3333-333333333333"),
|
||||
)
|
||||
|
||||
await engine.dispose()
|
||||
|
||||
assert [item.issue_no for item in matched] == ["MV-001"]
|
||||
assert unmatched == []
|
||||
@@ -1,25 +0,0 @@
|
||||
"""旧合同基础信息模块移除测试"""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def test_api_router_no_longer_registers_legacy_finance_contracts():
|
||||
router_source = (ROOT / "app" / "api" / "v1" / "router.py").read_text()
|
||||
|
||||
assert "finance_contracts" not in router_source
|
||||
assert "finance-contracts" not in router_source
|
||||
|
||||
|
||||
def test_active_backend_code_no_longer_imports_finance_contract_model():
|
||||
checked_paths = [
|
||||
ROOT / "app" / "db" / "base.py",
|
||||
ROOT / "app" / "crud" / "study.py",
|
||||
ROOT / "app" / "crud" / "site.py",
|
||||
ROOT / "app" / "crud" / "overview.py",
|
||||
]
|
||||
|
||||
for path in checked_paths:
|
||||
assert "FinanceContract" not in path.read_text()
|
||||
@@ -1,40 +0,0 @@
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
from app.models.user import User
|
||||
from app.services.site_contact_display import build_contact_display
|
||||
|
||||
|
||||
def make_user(user_id: uuid.UUID, full_name: str, email: str) -> User:
|
||||
return User(
|
||||
id=user_id,
|
||||
email=email,
|
||||
password_hash="hash",
|
||||
full_name=full_name,
|
||||
clinical_department="PMO",
|
||||
)
|
||||
|
||||
|
||||
def test_build_contact_display_resolves_comma_separated_user_ids():
|
||||
first_id = uuid.uuid4()
|
||||
second_id = uuid.uuid4()
|
||||
users = {
|
||||
first_id: make_user(first_id, "张三", "zhangsan@example.com"),
|
||||
second_id: make_user(second_id, "李四", "lisi@example.com"),
|
||||
}
|
||||
|
||||
display = build_contact_display(f"{first_id}, {second_id}", users)
|
||||
|
||||
assert display == "张三、李四"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("role", ["PM", "CRA", "PV", "QA", "CTA"])
|
||||
def test_build_contact_display_is_role_independent_for_site_read_roles(role: str):
|
||||
user_id = uuid.uuid4()
|
||||
users = {user_id: make_user(user_id, "周成", "zhoucheng@example.com")}
|
||||
|
||||
display = build_contact_display(str(user_id), users)
|
||||
|
||||
assert role
|
||||
assert display == "周成"
|
||||
@@ -1,54 +0,0 @@
|
||||
from datetime import date, datetime
|
||||
from types import SimpleNamespace
|
||||
import uuid
|
||||
|
||||
from app.crud.subject import _validate_actual_medication_count
|
||||
from app.schemas.subject import SubjectRead, SubjectUpdate
|
||||
|
||||
|
||||
def test_subject_update_accepts_actual_medication_count():
|
||||
payload = SubjectUpdate(actual_medication_count=12)
|
||||
|
||||
assert payload.actual_medication_count == 12
|
||||
|
||||
|
||||
def test_subject_update_accepts_screening_date():
|
||||
payload = SubjectUpdate(screening_date=date(2026, 5, 25))
|
||||
|
||||
assert payload.screening_date == date(2026, 5, 25)
|
||||
|
||||
|
||||
def test_subject_update_does_not_expose_status_input():
|
||||
assert "status" not in SubjectUpdate.model_fields
|
||||
|
||||
|
||||
def test_subject_read_includes_actual_medication_count():
|
||||
subject = SimpleNamespace(
|
||||
id=uuid.UUID("00000000-0000-0000-0000-000000000001"),
|
||||
study_id=uuid.UUID("00000000-0000-0000-0000-000000000002"),
|
||||
site_id=uuid.UUID("00000000-0000-0000-0000-000000000003"),
|
||||
subject_no="S001",
|
||||
status="ENROLLED",
|
||||
screening_date=None,
|
||||
consent_date=None,
|
||||
enrollment_date=None,
|
||||
baseline_date=None,
|
||||
completion_date=None,
|
||||
actual_medication_count=10,
|
||||
drop_reason=None,
|
||||
created_at=datetime(2026, 5, 9, 0, 0, 0),
|
||||
updated_at=datetime(2026, 5, 9, 0, 0, 0),
|
||||
)
|
||||
|
||||
data = SubjectRead.model_validate(subject)
|
||||
|
||||
assert data.actual_medication_count == 10
|
||||
|
||||
|
||||
def test_actual_medication_count_cannot_be_negative():
|
||||
try:
|
||||
_validate_actual_medication_count(-1)
|
||||
except ValueError as exc:
|
||||
assert "实际用药次数不能小于0" in str(exc)
|
||||
else:
|
||||
raise AssertionError("negative actual medication count should be rejected")
|
||||
Reference in New Issue
Block a user