ctms/backend/tests/test_api_permissions.py

"""单元测试：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_subject_history_permissions_are_labeled_as_medical_history_under_subjects():
    """病史记录权限应归入参与者管理，而不是独立的历史模块。"""
    expected_descriptions = {
        "subject_histories:create": "创建病史记录",
        "subject_histories:list": "查询病史记录列表",
        "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:list": "查询合同费用列表",
        "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_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:create",
        "fees_contracts_attachments:read",
        "fees_contracts_attachments:delete",
        "startup_initiation_attachments:create",
        "startup_initiation_attachments:read",
        "startup_initiation_attachments:delete",
        "startup_ethics_attachments:create",
        "startup_ethics_attachments:read",
        "startup_ethics_attachments:delete",
        "startup_auth_attachments:create",
        "startup_auth_attachments:read",
        "startup_auth_attachments:delete",
        "drug_shipments_attachments:create",
        "drug_shipments_attachments:read",
        "drug_shipments_attachments:delete",
        "precautions_attachments:create",
        "precautions_attachments:read",
        "precautions_attachments:delete",
        "faq_attachments:create",
        "faq_attachments:read",
        "faq_attachments:delete",
    }

    assert expected_keys <= set(API_ENDPOINT_PERMISSIONS)


def test_precautions_permissions_use_business_aligned_keys():
    """注意事项权限应使用业务一致的英文 key，并归入共享库父模块。"""
    expected_descriptions = {
        "precautions:create": "创建注意事项",
        "precautions:list": "查询注意事项列表",
        "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_use_module_permissions():
    """附件鉴权应使用模块附件权限，不再回退通用 attachments:*。"""
    attachments_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "attachments.py"
    source = attachments_path.read_text()

    expected_entity_modules = {
        '"contract_fee_contract": "fees_contracts_attachments"',
        '"contract_fee_voucher": "fees_contracts_attachments"',
        '"contract_fee_invoice": "fees_contracts_attachments"',
        '"startup_feasibility": "startup_initiation_attachments"',
        '"startup_ethics": "startup_ethics_attachments"',
        '"startup_kickoff": "startup_auth_attachments"',
        '"startup_kickoff_minutes": "startup_auth_attachments"',
        '"startup_kickoff_signin": "startup_auth_attachments"',
        '"startup_kickoff_ppt": "startup_auth_attachments"',
        '"training_authorization": "startup_auth_attachments"',
        '"drug_shipment": "drug_shipments_attachments"',
        '"precaution": "precautions_attachments"',
        '"faq_replies": "faq_attachments"',
    }

    for expected in expected_entity_modules:
        assert expected in source

    assert 'f"{module_prefix}:{operation}"' in source
    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 = {
        '"fees_contracts:read" if action == "read" else "fees_contracts:update"',
        'return f"startup_initiation:{parent_action}"',
        'return f"startup_ethics:{parent_action}"',
        '"startup_auth:read" if action == "read" else "startup_auth:update"',
        '"drug_shipments:read" if action == "read" else "drug_shipments:update"',
        '"precautions:read" if action == "read" else "precautions:update"',
        'return "faq:read"',
        'return "faq_reply:delete"',
        'return "faq_reply:create"',
    }

    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:list": "查询立项记录列表",
        "startup_initiation:read": "查询立项记录详情",
        "startup_initiation:update": "更新立项记录",
        "startup_initiation:delete": "删除立项记录",
        "startup_ethics:create": "创建伦理记录",
        "startup_ethics:list": "查询伦理记录列表",
        "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_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:list")' in summary_chunk
    assert 'require_api_permission("subject_aes:list")' 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:list": "查询参与者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:list": "查询监查访视问题列表",
        "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:list")' in list_chunk
    assert 'require_api_permission("monitoring_issues:list")' 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 '"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()

    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 delete_document_chunk
    assert "require_roles" not in delete_version_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_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)

    for role in configurable_roles:
        for endpoint_key, expected in payload[role].items():
            assert matrix[role][endpoint_key]["allowed"] is expected
            allowed = await role_has_api_permission(
                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