"""单元测试：API权限检查函数""" import pytest import uuid from pathlib import Path import ast from sqlalchemy import delete from sqlalchemy.ext.asyncio import AsyncSession from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, OPERATION_TO_ENDPOINTS, PROJECT_PERMISSION_ROLES, SYSTEM_PERMISSIONS from app.core.project_permissions import role_has_api_permission from app.core.project_permissions import get_api_endpoint_permissions, replace_api_endpoint_permissions from app.core.permission_cache import PermissionCache, set_permission_cache from app.core.permission_monitor import PermissionMonitor, set_permission_monitor from app.models.api_endpoint_permission import ApiEndpointPermission from app.models.study import Study from app.schemas.member import StudyMemberCreate @pytest.mark.asyncio async def test_api_permission_check_allowed(db_session: AsyncSession): """测试接口级权限检查 - 允许""" study_id = uuid.uuid4() # 创建权限记录 perm = ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="subjects:create", allowed=True, ) db_session.add(perm) await db_session.commit() # 验证权限（禁用前置权限检查） result = await role_has_api_permission( db_session, study_id, "CRA", "subjects:create", check_prerequisites=False ) assert result is True def test_all_backend_permission_guards_are_configurable(): """确保后端实际鉴权使用的 operation key 都能在权限管理中配置。""" api_dir = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" used_keys: set[str] = set() for path in api_dir.rglob("*.py"): source = path.read_text() tree = ast.parse(source) for node in ast.walk(tree): if not isinstance(node, ast.Call): continue if isinstance(node.func, ast.Name) and node.func.id == "require_api_permission": if node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str): used_keys.add(node.args[0].value) if isinstance(node.func, ast.Name) and node.func.id == "_ensure_project_access": if len(node.args) >= 4 and isinstance(node.args[3], ast.Constant) and isinstance(node.args[3].value, str): used_keys.add(node.args[3].value) assert used_keys assert used_keys <= set(API_ENDPOINT_PERMISSIONS) def test_read_permissions_do_not_duplicate_list_and_detail_entries(): """同一业务对象的列表和详情读取能力应在权限矩阵中合并为一个读取权限。""" merged_legacy_keys = { "subjects:list", "subject_aes:list", "fees_contracts:list", "project_members:list", "project_members:candidates", "sites:list", "startup_ethics:list", "startup_initiation:list", "monitoring_issues:list", "drug_shipments:list", "audit_logs:list", "precautions:list", "subject_histories:list", "material_equipments:list", } assert merged_legacy_keys.isdisjoint(API_ENDPOINT_PERMISSIONS) expected_read_keys = { "subjects:read", "subject_aes:read", "fees_contracts:read", "project_members:read", "sites:read", "startup_ethics:read", "startup_initiation:read", "monitoring_issues:read", "drug_shipments:read", "audit_logs:read", "precautions:read", "subject_histories:read", "material_equipments:read", } assert expected_read_keys <= set(API_ENDPOINT_PERMISSIONS) def test_site_contact_display_source_permissions_cover_non_pm_roles(): """负责人展示名来自中心列表，读取角色不应局限于 PM。""" sites_read_roles = set(API_ENDPOINT_PERMISSIONS["sites:read"]["default_roles"]) startup_auth_read_roles = set(API_ENDPOINT_PERMISSIONS["startup_auth:read"]["default_roles"]) assert {"PM", "CRA", "PV", "QA", "CTA"} <= sites_read_roles assert {"PM", "CRA", "PV", "CTA"} <= startup_auth_read_roles @pytest.mark.asyncio async def test_legacy_read_permission_overrides_are_folded_into_canonical_read_key(db_session: AsyncSession): """历史项目权限中的列表读取 key 应继续作用到合并后的读取权限。""" study_id = uuid.uuid4() db_session.add_all( ( ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="subjects:list", allowed=False, ), ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="project_members:candidates", allowed=True, ), ) ) await db_session.commit() matrix = await get_api_endpoint_permissions(db_session, study_id) assert "subjects:list" not in matrix["CRA"] assert matrix["CRA"]["subjects:read"]["allowed"] is False assert "project_members:candidates" not in matrix["CRA"] assert matrix["CRA"]["project_members:read"]["allowed"] is True def test_backend_read_routes_use_canonical_read_permission_keys(): """列表接口和详情接口应共用合并后的读取权限 key。""" api_dir = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" merged_legacy_keys = { "subjects:list", "subject_aes:list", "fees_contracts:list", "project_members:list", "project_members:candidates", "sites:list", "startup_ethics:list", "startup_initiation:list", "monitoring_issues:list", "drug_shipments:list", "audit_logs:list", "precautions:list", "subject_histories:list", "material_equipments:list", } source = "\n".join(path.read_text() for path in api_dir.rglob("*.py")) for endpoint_key in merged_legacy_keys: assert f'require_api_permission("{endpoint_key}")' not in source assert f'_ensure_study_access(db, study_id, current_user, "{endpoint_key}")' not in source def test_subject_history_permissions_are_labeled_as_medical_history_under_subjects(): """病史记录权限应归入参与者管理，而不是独立的历史模块。""" expected_descriptions = { "subject_histories:create": "创建病史记录", "subject_histories:read": "查询病史记录", "subject_histories:update": "更新病史记录", "subject_histories:delete": "删除病史记录", } for endpoint_key, description in expected_descriptions.items(): assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "subjects" assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description def test_contract_fee_permissions_do_not_include_legacy_finance_contracts(): """旧合同基础信息权限应从当前权限矩阵移除。""" for endpoint_key in API_ENDPOINT_PERMISSIONS: assert not endpoint_key.startswith("finance_contracts:") assert not endpoint_key.startswith("fees_payments:") assert not endpoint_key.startswith("fees_attachments:") expected_descriptions = { "fees_contracts:create": "创建合同费用", "fees_contracts:read": "查询合同费用", "fees_contracts:update": "更新合同费用", "fees_contracts:delete": "删除合同费用", } for endpoint_key, description in expected_descriptions.items(): assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "fees" assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description def test_setup_config_delete_version_permission_is_labeled_as_delete(): """删除立项配置版本应在权限矩阵中展示为删除，而不是写入。""" permission = API_ENDPOINT_PERMISSIONS["setup_config:delete_version"] assert permission["module"] == "setup_config" assert permission["action"] == "delete" assert permission["description"] == "删除立项配置版本" def test_generic_attachment_permissions_are_removed_from_matrix(): """前端和模板迁移后，项目权限矩阵不再暴露通用 attachments:*。""" assert not any(endpoint_key.startswith("attachments:") for endpoint_key in API_ENDPOINT_PERMISSIONS) def test_module_attachment_permissions_are_available(): """附件权限矩阵只保留独立删除项，创建与读取由业务主记录权限控制。""" expected_keys = { "fees_contracts_attachments:delete", "startup_initiation_attachments:delete", "startup_ethics_attachments:delete", "startup_auth_attachments:delete", "drug_shipments_attachments:delete", "material_equipments_attachments:delete", "precautions_attachments:delete", "faq_attachments:delete", } assert expected_keys <= set(API_ENDPOINT_PERMISSIONS) removed_action_keys = { f"{prefix}:{action}" for prefix in ( "fees_contracts_attachments", "startup_initiation_attachments", "startup_ethics_attachments", "startup_auth_attachments", "drug_shipments_attachments", "material_equipments_attachments", "precautions_attachments", "faq_attachments", ) for action in ("create", "read") } assert removed_action_keys.isdisjoint(API_ENDPOINT_PERMISSIONS) def test_precautions_permissions_use_business_aligned_keys(): """注意事项权限应使用业务一致的英文 key，并归入共享库父模块。""" expected_descriptions = { "precautions:create": "创建注意事项", "precautions:read": "查询注意事项", "precautions:update": "更新注意事项", "precautions:delete": "删除注意事项", } for endpoint_key, description in expected_descriptions.items(): assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "shared_library" assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description assert not any(key.startswith("knowledge_notes:") for key in API_ENDPOINT_PERMISSIONS) assert not any(key.startswith("knowledge_notes_attachments:") for key in API_ENDPOINT_PERMISSIONS) def test_faq_and_precautions_are_shared_library_sibling_sections(): """医学咨询与注意事项应同属共享库，由前端模块细分区分。""" shared_library_keys = OPERATION_TO_ENDPOINTS["shared_library"] assert "faq:read" in shared_library_keys["read"] assert "precautions:read" in shared_library_keys["read"] assert "faq:create" in shared_library_keys["write"] assert "precautions:create" in shared_library_keys["write"] assert API_ENDPOINT_PERMISSIONS["faq:read"]["module"] == "shared_library" assert API_ENDPOINT_PERMISSIONS["precautions:read"]["module"] == "shared_library" def test_precautions_runtime_code_uses_business_entity_names(): """运行时代码不应继续使用 knowledge_note 作为注意事项实体命名。""" backend_root = Path(__file__).resolve().parents[1] / "app" ignored = {"__pycache__"} combined_source = "\n".join( path.read_text(encoding="utf-8") for path in backend_root.rglob("*.py") if not ignored.intersection(path.parts) ) assert "KnowledgeNote" not in combined_source assert "knowledge_note" not in combined_source assert "knowledge_notes" not in combined_source assert "Precaution" in combined_source def test_legacy_fee_attachment_router_is_not_registered(): """费用附件应统一走通用附件 API，不再注册专用 /fees/attachments 路由。""" router_source = (Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "router.py").read_text() assert "fees_attachments" not in router_source def test_legacy_fee_attachment_runtime_code_is_removed(): """旧费用附件运行时代码应删除，避免绕过模块附件权限的死代码被重新注册。""" backend_root = Path(__file__).resolve().parents[1] / "app" legacy_tokens = { "fees_attachments", "fee_attachment", "FeeAttachment", "fee_attachments", } offenders: list[str] = [] for path in backend_root.rglob("*.py"): source = path.read_text(encoding="utf-8") if any(token in source or token in path.name for token in legacy_tokens): offenders.append(str(path.relative_to(backend_root))) assert offenders == [] def test_attachment_permissions_do_not_fall_back_to_generic_permissions(): """附件鉴权不回退通用 attachments:*，业务附件由父业务权限或删除附件权限控制。""" attachments_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "attachments.py" source = attachments_path.read_text() assert 'f"attachments:{operation}"' not in source assert "role_has_api_permission(" in source assert "parent_permission" in source assert source.count("_ensure_attachment_permission(") >= 8 def test_attachment_parent_permissions_cover_business_modules(): """已知附件实体必须叠加对应父模块权限，避免只靠附件权限越权读取业务数据。""" attachments_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "attachments.py" source = attachments_path.read_text() expected_parent_permissions = { '"create": "fees_contracts:create"', '"read": "fees_contracts:read"', '"delete": "fees_contracts_attachments:delete"', '"create": "startup_initiation:create"', '"read": "startup_initiation:read"', '"delete": "startup_initiation_attachments:delete"', '"create": "startup_ethics:create"', '"read": "startup_ethics:read"', '"delete": "startup_ethics_attachments:delete"', '"create": "startup_auth:create"', '"read": "startup_auth:read"', '"delete": "startup_auth_attachments:delete"', '"create": "drug_shipments:create"', '"read": "drug_shipments:read"', '"delete": "drug_shipments_attachments:delete"', '"create": "material_equipments:update"', '"read": "material_equipments:read"', '"delete": "material_equipments_attachments:delete"', '"create": "precautions:create"', '"read": "precautions:read"', '"delete": "precautions_attachments:delete"', '"create": "faq_reply:create"', '"read": "faq:read"', '"delete": "faq_attachments:delete"', } for expected in expected_parent_permissions: assert expected in source def test_startup_ethics_permissions_use_visible_business_language(): """立项与伦理权限矩阵应使用前台一致的业务名称。""" expected_descriptions = { "startup_initiation:create": "创建立项记录", "startup_initiation:read": "查询立项记录", "startup_initiation:update": "更新立项记录", "startup_initiation:delete": "删除立项记录", "startup_ethics:create": "创建伦理记录", "startup_ethics:read": "查询伦理记录", "startup_ethics:update": "更新伦理记录", "startup_ethics:delete": "删除伦理记录", } assert not any(key.startswith("feasibility:") for key in API_ENDPOINT_PERMISSIONS) assert not any(key.startswith("ethics:") for key in API_ENDPOINT_PERMISSIONS) for endpoint_key, description in expected_descriptions.items(): assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "startup_ethics" assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description def test_startup_auth_permissions_only_include_wired_operations(): """启动与授权权限矩阵只应展示实际路由使用的操作。""" stale_keys = { "budget:create", "budget:list", "budget:read", "budget:update", "budget:delete", "timeline:create", "timeline:list", "timeline:read", "timeline:update", "timeline:delete", } assert stale_keys.isdisjoint(API_ENDPOINT_PERMISSIONS) for endpoint_keys in OPERATION_TO_ENDPOINTS["startup_auth"].values(): assert stale_keys.isdisjoint(endpoint_keys) def test_training_authorization_list_can_filter_by_site_name(): """启动会详情页查询培训授权时应支持按当前中心过滤，避免串中心显示。""" route_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "startup.py" crud_path = Path(__file__).resolve().parents[1] / "app" / "crud" / "startup.py" route_source = route_path.read_text() crud_source = crud_path.read_text() assert "site_name: str | None = None" in route_source assert "site_names=site_names, site_name=site_name" in route_source assert "site_name: str | None = None" in crud_source assert "TrainingAuthorization.site_name == site_name" in crud_source def test_site_creation_initializes_all_startup_baseline_records(): """创建中心时应生成启动模块默认记录，但后续手动删除不应依赖列表页重建。""" route_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "sites.py" source = route_path.read_text() assert "StartupFeasibilityCreate(site_id=site.id)" in source assert "StartupEthicsCreate(site_id=site.id)" in source assert "KickoffMeetingCreate(site_id=site.id)" in source assert "await startup_crud.create_kickoff(" in source def test_visit_list_permission_matches_wired_visit_endpoints(): """访视当前没有详情查询接口，列表接口应只使用 visits:list。""" visits_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "visits.py" source = visits_path.read_text() assert "visits:list" in API_ENDPOINT_PERMISSIONS assert "visits:read" not in API_ENDPOINT_PERMISSIONS assert 'require_api_permission("visits:list")' in source assert 'require_api_permission("visits:read")' not in source def test_ae_subject_and_summary_share_read_permission(): """风险问题AE汇总与参与者AE列表应使用同一个读取权限。""" aes_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "aes.py" source = aes_path.read_text() summary_chunk = source[source.index('"/summary"') : source.index('@router.get(\n "/"')] list_chunk = source[source.index('@router.get(\n "/",') : source.index('@router.get(\n "/{ae_id}"')] detail_chunk = source[source.index('response_model=AERead,\n dependencies=[Depends(require_api_permission("subject_aes:read"))]') : source.index('@router.patch')] assert 'require_api_permission("subject_aes:read")' in summary_chunk assert 'require_api_permission("subject_aes:read")' in list_chunk assert 'require_api_permission("subject_aes:read")' in detail_chunk def test_ae_mutation_permissions_are_under_subject_management(): """AE维护发生在参与者详情内，风险问题模块只保留汇总读取权限。""" expected_descriptions = { "subject_aes:create": "创建参与者AE", "subject_aes:read": "查询参与者AE", "subject_aes:update": "更新参与者AE", "subject_aes:delete": "删除参与者AE", } for endpoint_key, description in expected_descriptions.items(): assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "subjects" assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description for stale_key in ("risk_issues:create", "risk_issues:list", "risk_issues:read", "risk_issues:update", "risk_issues:delete"): assert stale_key not in API_ENDPOINT_PERMISSIONS for stale_key in ("risk_issue_aes:list", "risk_issue_pds:list"): assert stale_key not in API_ENDPOINT_PERMISSIONS def test_risk_issue_pd_summary_uses_subject_pd_permission(): """风险问题PD汇总与参与者PD列表应使用同一个读取权限。""" pds_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "study_subject_pds.py" source = pds_path.read_text() assert 'require_api_permission("subject_pds:list")' in source assert 'require_api_permission("risk_issues:list")' not in source def test_risk_issue_monitoring_visit_uses_monitoring_issue_permissions(): """监查访视问题是风险问题下唯一独立权限子模块。""" monitoring_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "monitoring_visit_issues.py" source = monitoring_path.read_text() legacy_issue_prefix = "monitoring" + "_issues" tree = ast.parse(source) used_permission_keys = { node.args[0].value for node in ast.walk(tree) if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "require_api_permission" and node.args and isinstance(node.args[0], ast.Constant) and isinstance(node.args[0].value, str) } list_chunk = source[source.index('@router.get(\n "/issues"') : source.index("@router.post")] create_chunk = source[source.index("@router.post") : source.index('@router.get(\n "/issues/export"')] export_chunk = source[source.index('@router.get(\n "/issues/export"') : source.index('@router.get(\n "/issues/{issue_id}"')] detail_chunk = source[source.index('@router.get(\n "/issues/{issue_id}"') : source.index("@router.patch")] update_chunk = source[source.index("@router.patch") : source.index("@router.delete")] delete_chunk = source[source.index("@router.delete") : source.index('@router.post(\n "/issues/import"')] import_chunk = source[source.index('@router.post(\n "/issues/import"') :] expected_descriptions = { "monitoring_issues:read": "查询监查访视问题", "monitoring_issues:create": "创建监查访视问题", "monitoring_issues:update": "更新监查访视问题", "monitoring_issues:delete": "删除监查访视问题", } for endpoint_key, description in expected_descriptions.items(): assert API_ENDPOINT_PERMISSIONS[endpoint_key]["module"] == "risk_issues" assert API_ENDPOINT_PERMISSIONS[endpoint_key]["description"] == description assert 'require_api_permission("monitoring_issues:read")' in list_chunk assert 'require_api_permission("monitoring_issues:read")' in export_chunk assert 'require_api_permission("monitoring_issues:read")' in detail_chunk assert 'require_api_permission("monitoring_issues:create")' in create_chunk assert 'require_api_permission("monitoring_issues:create")' in import_chunk assert 'require_api_permission("monitoring_issues:update")' in update_chunk assert 'require_api_permission("monitoring_issues:delete")' in delete_chunk assert used_permission_keys == set(expected_descriptions) for stale_key in ( f"{legacy_issue_prefix}:close", f"{legacy_issue_prefix}:history", ): assert stale_key not in API_ENDPOINT_PERMISSIONS assert not any(config["module"] == "monitoring" + "_audit" for config in API_ENDPOINT_PERMISSIONS.values()) def test_document_service_uses_specific_document_permission_keys(): """文档模块应使用 create/update/delete 细粒度权限，而不是全部退化为 update。""" service_path = Path(__file__).resolve().parents[1] / "app" / "services" / "document_service.py" source = service_path.read_text() assert '"create_document": "documents:create"' in source assert '"update_document": "documents:update"' in source assert '"create_version": "documents:update"' in source assert '"distribute": "documents:update"' in source assert '"delete_document": "documents:delete"' in source assert 'else "documents:update"' not in source assert "await _ensure_study_access(db, doc.trial_id, current_user, action=\"distribute\")" in source def test_document_routes_do_not_hardcode_admin_for_matrix_controlled_actions(): """文档删除和版本删除应由项目权限矩阵控制，而不是路由层硬编码 ADMIN。""" route_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "documents.py" source = route_path.read_text() update_document_chunk = source[source.index("async def update_document") : source.index("@router.delete", source.index("async def update_document"))] delete_document_chunk = source[source.index("async def delete_document") : source.index("@router.post", source.index("async def delete_document"))] delete_version_chunk = source[source.index("async def delete_version") : source.index("@router.post", source.index("async def delete_version"))] assert "require_roles" not in update_document_chunk assert "require_roles" not in delete_document_chunk assert "require_roles" not in delete_version_chunk assert "仅管理员可编辑文档" not in update_document_chunk assert "仅管理员可删除文档" not in delete_document_chunk assert "仅管理员可删除版本" not in delete_version_chunk def test_project_permission_config_is_system_level_permission(): """项目权限配置应归属系统级权限，不能作为项目矩阵里的自管理权限。""" assert "permissions:read" not in API_ENDPOINT_PERMISSIONS assert "permissions:update" not in API_ENDPOINT_PERMISSIONS assert SYSTEM_PERMISSIONS["system:permissions:project_config"]["description"] == "配置项目接口权限" assert "PM" in SYSTEM_PERMISSIONS["system:permissions:project_config"]["roles"] def test_project_api_permission_routes_use_system_project_config_permission(): """项目权限矩阵 API 应明确依赖 system:permissions:project_config。""" route_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "api_permissions.py" source = route_path.read_text() get_chunk = source[source.index("async def get_study_api_permissions") : source.index("@study_router.put")] update_chunk = source[source.index("async def update_study_api_permissions") :] assert 'require_system_permission("system:permissions:project_config")' in get_chunk assert 'require_system_permission("system:permissions:project_config")' in update_chunk assert 'require_study_roles(["PM"])' not in get_chunk assert 'require_study_roles(["PM"])' not in update_chunk def test_audit_export_event_uses_export_permission(): """审计导出事件应使用 audit_logs:export，而不是仅要求项目成员。""" route_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "audit_logs.py" source = route_path.read_text() event_chunk = source[source.index("async def create_audit_event") :] assert 'require_api_permission("audit_logs:export")' in event_chunk assert "require_study_member()" not in event_chunk def test_audit_export_event_does_not_trust_client_detail(): """导出审计事件的业务描述应由后端生成，不能信任客户端 detail。""" route_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "audit_logs.py" source = route_path.read_text() event_chunk = source[source.index("async def create_audit_event") :] assert "detail=payload.detail" not in event_chunk assert "entity_type=payload.entity_type" not in event_chunk assert "entity_id=payload.entity_id" not in event_chunk assert "_build_export_audit_detail" in event_chunk def test_audit_logs_do_not_expose_delete_endpoint(): """审计日志不能提供单条删除接口，避免破坏审计不可篡改性。""" route_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "audit_logs.py" source = route_path.read_text() assert "@router.delete" not in source assert "delete_audit_log" not in source def test_etmf_audit_details_are_structured_json(): """eTMF 审计详情不能手拼 JSON 或写空对象。""" service_path = Path(__file__).resolve().parents[1] / "app" / "services" / "etmf_service.py" source = service_path.read_text() assert "json.dumps" in source assert "detail=f'{{" not in source assert 'detail="{}"' not in source def test_setup_config_diff_summary_keeps_all_business_changes_and_skips_ids(): """立项配置审计明细应完整保留业务字段，跳过行内技术 ID。""" route_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "studies.py" source = route_path.read_text() summary_chunk = source[source.index("def _setup_collect_diff_lines") : source.index("@router.post")] assert '_SETUP_TECHNICAL_FIELDS = {"id"}' in source assert "if key in _SETUP_TECHNICAL_FIELDS:" in summary_chunk assert "已省略" not in summary_chunk assert "max_lines" not in summary_chunk def test_subject_audit_uses_structured_business_snapshot(): """参与者审计应把编号放入对象，把业务变化放入详情。""" route_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "subjects.py" source = route_path.read_text() create_subject_source = source[source.index("async def create_subject") : source.index("@router.get")] update_subject_source = source[source.index("async def update_subject") : source.index("@router.delete")] delete_subject_source = source[source.index("async def delete_subject") :] assert "_subject_audit_snapshot" in source assert '"targetName": subject.subject_no' in create_subject_source assert '"description": f"创建参与者 {subject.subject_no}"' in create_subject_source assert '"targetName": updated.subject_no' in update_subject_source assert '"description": description' in update_subject_source assert '"targetName": target_name' in delete_subject_source assert '"description": f"删除参与者 {target_name}"' in delete_subject_source assert '"before": before_snapshot' in delete_subject_source assert 'detail=f"参与者 {subject.subject_no} 已创建"' not in source assert 'f"参与者 {updated.subject_no} 状态' not in source assert 'detail=f"参与者 {subject_id} 已删除"' not in delete_subject_source def test_faq_audit_uses_business_descriptions(): """医学咨询审计详情应使用业务描述，避免 FAQ 技术文案。""" api_dir = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" category_source = (api_dir / "faq_categories.py").read_text() item_source = (api_dir / "faqs.py").read_text() assert 'description": f"创建“{category.name}”分类"' in category_source assert 'description": f"更新“{updated.name}”分类"' in category_source assert 'description": f"删除“{category_name}”分类"' in category_source assert 'f"创建医学咨询问题“{question_name}”"' in item_source assert 'f"更新医学咨询问题“{question_name}”"' in item_source assert 'f"回复医学咨询问题“{question_name}”"' in item_source assert 'f"删除医学咨询问题“{question_name}”"' in item_source assert 'f"删除医学咨询问题“{question_name}”的回复"' in item_source assert "FAQ 分类 {category.name} 已创建" not in category_source assert 'detail="FAQ 已创建"' not in item_source assert 'detail = "FAQ updated"' not in item_source assert '"description": "创建医学咨询问题"' not in item_source def test_domain_audits_use_business_object_names(): """药品、设备、访视、AE、费用、立项等审计应记录具体业务对象。""" api_dir = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" sources = { name: (api_dir / name).read_text() for name in [ "drug_shipments.py", "material_equipments.py", "visits.py", "aes.py", "precautions.py", "fees_contracts.py", "startup.py", "subject_histories.py", ] } assert "_shipment_audit_description" in sources["drug_shipments.py"] assert "_equipment_audit_detail" in sources["material_equipments.py"] assert "_visit_audit_detail" in sources["visits.py"] assert "_ae_audit_detail" in sources["aes.py"] assert "_precaution_audit_detail" in sources["precautions.py"] assert "_contract_audit_detail" in sources["fees_contracts.py"] assert "_payment_audit_detail" in sources["fees_contracts.py"] assert "_feasibility_audit_detail" in sources["startup.py"] assert "_ethics_audit_detail" in sources["startup.py"] assert "_kickoff_audit_detail" in sources["startup.py"] assert "_training_audit_detail" in sources["startup.py"] assert "_history_audit_detail" in sources["subject_histories.py"] forbidden_fragments = [ 'detail=f"访视 {visit_id} 已删除"', 'detail=f"AE {ae.id} 已创建"', 'detail=f"AE {ae_id} 已删除"', 'detail=f"设备 {item.id} 已创建"', 'detail=f"设备 {equipment_id} 已更新"', 'detail=f"设备 {equipment_id} 已删除"', 'detail=f"注意事项 {precaution_id} 已更新"', 'detail=f"注意事项 {precaution_id} 已删除"', 'detail="合同费用已创建"', 'detail="合同费用已更新"', 'detail="合同费用已删除"', 'detail="合同费用分期已创建"', 'detail="合同费用分期已更新"', 'detail="合同费用分期已删除"', 'detail="立项记录已创建"', 'detail="立项记录已更新"', 'detail="立项记录已删除"', 'detail="伦理记录已创建"', 'detail="伦理记录已更新"', 'detail="伦理记录已删除"', 'detail="启动会记录已创建"', 'detail="启动会记录已更新"', 'detail="培训授权人员已创建"', 'detail="培训授权人员已更新"', 'detail="培训授权人员已删除"', 'detail="病史记录已创建"', 'detail="病史记录已更新"', 'detail="病史记录已删除"', ] combined_source = "\n".join(sources.values()) for fragment in forbidden_fragments: assert fragment not in combined_source def test_user_role_contains_current_project_roles(): assert {"QA", "CTA"} <= set(PROJECT_PERMISSION_ROLES) @pytest.mark.asyncio async def test_default_matrix_covers_every_role_and_permission(db_session: AsyncSession): """逐一验证预设项目权限角色在每个权限上的默认矩阵。""" study_id = uuid.uuid4() matrix = await get_api_endpoint_permissions(db_session, study_id) assert {"QA", "CTA"} <= set(PROJECT_PERMISSION_ROLES) assert any("QA" in config["default_roles"] for config in API_ENDPOINT_PERMISSIONS.values()) assert any("CTA" in config["default_roles"] for config in API_ENDPOINT_PERMISSIONS.values()) assert set(matrix) == set(PROJECT_PERMISSION_ROLES) for role in PROJECT_PERMISSION_ROLES: assert set(matrix[role]) == set(API_ENDPOINT_PERMISSIONS) for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items(): 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): """逐一验证前端提交格式会落库，并被后端鉴权函数按相同结果读取。""" cache = PermissionCache() monitor = PermissionMonitor() set_permission_cache(cache) 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(configurable_roles) } matrix = await replace_api_endpoint_permissions( db_session, study_id, payload, confirm_prerequisite_adjustments=True, ) for role in configurable_roles: for endpoint_key in payload[role].keys(): expected = matrix[role][endpoint_key]["allowed"] assert matrix[role][endpoint_key]["allowed"] is expected allowed = await role_has_api_permission( db_session, study_id, role, endpoint_key, check_prerequisites=False, ) assert allowed is expected @pytest.mark.asyncio async def test_custom_active_role_permissions_take_effect(db_session: AsyncSession): """验证项目自定义角色配置权限后能被后端鉴权读取。""" study_id = uuid.uuid4() db_session.add( Study( id=study_id, code=f"CUSTOM-{study_id.hex[:8]}", name="自定义角色权限测试", status="ACTIVE", is_locked=False, visit_schedule=[], active_roles=["DATA_MANAGER"], ) ) await db_session.commit() matrix = await get_api_endpoint_permissions(db_session, study_id) assert "DATA_MANAGER" in matrix assert matrix["DATA_MANAGER"]["subjects:read"]["allowed"] is False await replace_api_endpoint_permissions( db_session, study_id, {"DATA_MANAGER": {"subjects:read": True, "subjects:update": False}}, ) assert await role_has_api_permission( db_session, study_id, "DATA_MANAGER", "subjects:read", check_prerequisites=False, ) is True assert await role_has_api_permission( db_session, study_id, "DATA_MANAGER", "subjects:update", check_prerequisites=False, ) is False @pytest.mark.asyncio async def test_permission_matrix_filters_stale_unknown_endpoint_overrides(db_session: AsyncSession): """数据库历史残留的未知权限项不应重新出现在权限矩阵中。""" study_id = uuid.uuid4() db_session.add_all( ( ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="subject_history:export", allowed=True, ), ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="subject_pds:read", allowed=True, ), ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="visits:read", allowed=True, ), ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="risk_issues:create", allowed=True, ), ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="risk_issues:list", allowed=True, ), ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="risk_issue_aes:list", allowed=True, ), ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key=("monitoring" + "_issues") + ":history", allowed=True, ), ) ) await db_session.commit() matrix = await get_api_endpoint_permissions(db_session, study_id) assert "subject_history:export" not in matrix["CRA"] assert "subject_pds:read" not in matrix["CRA"] assert "visits:read" not in matrix["CRA"] assert "risk_issues:create" not in matrix["CRA"] assert "risk_issues:list" not in matrix["CRA"] assert "risk_issue_aes:list" not in matrix["CRA"] assert ("monitoring" + "_issues") + ":history" not in matrix["CRA"] def test_admin_cannot_be_used_as_project_custom_role(): """避免项目角色 ADMIN 触发后端系统管理员绕过逻辑。""" with pytest.raises(ValueError): StudyMemberCreate(user_id=uuid.uuid4(), role_in_study="ADMIN") @pytest.mark.asyncio async def test_api_permission_check_denied(db_session: AsyncSession): """测试接口级权限检查 - 拒绝""" study_id = uuid.uuid4() # 创建权限记录 perm = ApiEndpointPermission( study_id=study_id, role="PV", endpoint_key="subjects:create", allowed=False, ) db_session.add(perm) await db_session.commit() # 验证权限（禁用前置权限检查） result = await role_has_api_permission( db_session, study_id, "PV", "subjects:create", check_prerequisites=False ) assert result is False @pytest.mark.asyncio async def test_admin_always_allowed(db_session: AsyncSession): """测试ADMIN角色总是被允许""" study_id = uuid.uuid4() result = await role_has_api_permission( db_session, study_id, "ADMIN", "subjects:create" ) assert result is True @pytest.mark.asyncio async def test_api_permission_read_endpoint(db_session: AsyncSession): """测试读取端点权限""" study_id = uuid.uuid4() # 创建读取权限 perm = ApiEndpointPermission( study_id=study_id, role="PV", endpoint_key="subjects:read", allowed=True, ) db_session.add(perm) await db_session.commit() # 验证权限（禁用前置权限检查） result = await role_has_api_permission( db_session, study_id, "PV", "subjects:read", check_prerequisites=False ) assert result is True @pytest.mark.asyncio async def test_api_permission_different_endpoints(db_session: AsyncSession): """测试不同端点的权限独立""" study_id = uuid.uuid4() # 创建权限：允许GET，拒绝POST get_perm = ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="subjects:read", allowed=True, ) post_perm = ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="subjects:create", allowed=False, ) db_session.add(get_perm) db_session.add(post_perm) await db_session.commit() # 验证权限（禁用前置权限检查） get_result = await role_has_api_permission( db_session, study_id, "CRA", "subjects:read", check_prerequisites=False ) post_result = await role_has_api_permission( db_session, study_id, "CRA", "subjects:create", check_prerequisites=False ) assert get_result is True assert post_result is False @pytest.mark.asyncio async def test_api_permission_different_roles(db_session: AsyncSession): """测试不同角色的权限独立""" study_id = uuid.uuid4() # 创建权限：CRA允许，PV拒绝 cra_perm = ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="subjects:create", allowed=True, ) pv_perm = ApiEndpointPermission( study_id=study_id, role="PV", endpoint_key="subjects:create", allowed=False, ) db_session.add(cra_perm) db_session.add(pv_perm) await db_session.commit() # 验证权限（禁用前置权限检查） cra_result = await role_has_api_permission( db_session, study_id, "CRA", "subjects:create", check_prerequisites=False ) pv_result = await role_has_api_permission( db_session, study_id, "PV", "subjects:create", check_prerequisites=False ) assert cra_result is True assert pv_result is False @pytest.mark.asyncio async def test_api_permission_different_studies(db_session: AsyncSession): """测试不同项目的权限独立""" study_id_1 = uuid.uuid4() study_id_2 = uuid.uuid4() # 创建权限：项目1允许，项目2拒绝 perm_1 = ApiEndpointPermission( study_id=study_id_1, role="CRA", endpoint_key="subjects:create", allowed=True, ) perm_2 = ApiEndpointPermission( study_id=study_id_2, role="CRA", endpoint_key="subjects:create", allowed=False, ) db_session.add(perm_1) db_session.add(perm_2) await db_session.commit() # 验证权限（禁用前置权限检查） result_1 = await role_has_api_permission( db_session, study_id_1, "CRA", "subjects:create", check_prerequisites=False ) result_2 = await role_has_api_permission( db_session, study_id_2, "CRA", "subjects:create", check_prerequisites=False ) assert result_1 is True assert result_2 is False @pytest.mark.asyncio async def test_api_permission_none_role(db_session: AsyncSession): """测试None角色的权限检查""" study_id = uuid.uuid4() result = await role_has_api_permission( db_session, study_id, None, "subjects:create", check_prerequisites=False ) assert result is False @pytest.mark.asyncio async def test_api_permission_unknown_endpoint(db_session: AsyncSession): """测试未知端点的权限检查""" study_id = uuid.uuid4() result = await role_has_api_permission( db_session, study_id, "CRA", "unknown:endpoint", check_prerequisites=False ) assert result is False @pytest.mark.asyncio async def test_api_permission_unknown_endpoint_ignores_stale_override(db_session: AsyncSession): """即使数据库残留未知权限项，鉴权入口也不应放行。""" study_id = uuid.uuid4() db_session.add_all( ( ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="subject_history:export", allowed=True, ), ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="subject_pds:read", allowed=True, ), ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="risk_issues:list", allowed=True, ), ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="risk_issue_aes:list", allowed=True, ), ) ) await db_session.commit() history_result = await role_has_api_permission( db_session, study_id, "CRA", "subject_history:export", check_prerequisites=False ) pds_result = await role_has_api_permission( db_session, study_id, "CRA", "subject_pds:read", check_prerequisites=False ) risk_issue_result = await role_has_api_permission( db_session, study_id, "CRA", "risk_issues:list", check_prerequisites=False ) risk_issue_ae_result = await role_has_api_permission( db_session, study_id, "CRA", "risk_issue_aes:list", check_prerequisites=False ) assert history_result is False assert pds_result is False assert risk_issue_result is False assert risk_issue_ae_result is False @pytest.mark.asyncio async def test_api_permission_check_uses_project_permission_cache(db_session: AsyncSession): """测试接口权限检查会复用项目权限缓存并记录命中指标""" cache = PermissionCache() monitor = PermissionMonitor() set_permission_cache(cache) set_permission_monitor(monitor) study_id = uuid.uuid4() db_session.add( ApiEndpointPermission( study_id=study_id, role="CRA", endpoint_key="subjects:create", allowed=True, ) ) await db_session.commit() first_result = await role_has_api_permission( db_session, study_id, "CRA", "subjects:create", check_prerequisites=False ) await db_session.execute( delete(ApiEndpointPermission).where(ApiEndpointPermission.study_id == study_id) ) await db_session.commit() second_result = await role_has_api_permission( db_session, study_id, "CRA", "subjects:create", check_prerequisites=False ) cache_metrics = monitor.get_metrics()["cache_metrics"] assert first_result is True assert second_result is True assert cache.get_cache_stats()["project_permissions_count"] == 1 assert cache_metrics["cache_misses"] == 1 assert cache_metrics["cache_hits"] == 1 @pytest.mark.asyncio async def test_replace_api_endpoint_permissions_invalidates_project_permission_cache( db_session: AsyncSession, ): """测试替换权限后会失效旧缓存并记录失效指标""" from app.core.project_permissions import replace_api_endpoint_permissions cache = PermissionCache() monitor = PermissionMonitor() set_permission_cache(cache) set_permission_monitor(monitor) study_id = uuid.uuid4() await replace_api_endpoint_permissions( db_session, study_id, {"CRA": {"subjects:create": True}}, ) assert cache.get_cache_stats()["project_permissions_count"] == 1 await replace_api_endpoint_permissions( db_session, study_id, {"CRA": {"subjects:create": False}}, ) result = await role_has_api_permission( db_session, study_id, "CRA", "subjects:create", check_prerequisites=False ) cache_metrics = monitor.get_metrics()["cache_metrics"] assert result is False assert cache_metrics["cache_invalidations"] == 2