zkxw/backend/tests/test_web_sensitive_permissions.py

import asyncio
from pathlib import Path

import pytest
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.core import menu_guard
from app.core.auth import get_current_user
from app.core.database import get_db
from app.models.base import Base
from app.models.models import (
    SysMenu,
    SysPost,
    SysPostRole,
    SysRole,
    SysRoleMenu,
    SysUser,
    SysUserPost,
    SysUserRole,
)


PROTOCOL_LIST = "biz:cellStorageProtocol:list"
PROTOCOL_AUDIT = "biz:cellStorageProtocolAudit:list"
PAYMENT_CONFIRM = "biz:cellStoragePaymentConfirm:list"
REFUND_CONFIRM = "biz:cellStorageRefundConfirm:list"
STORAGE_CERT = "cell-storage:storage-cert:list"
QC_MENUS = {"biz:qcWorkOrder:list", "biz:myQcTask:list"}
STORAGE_CERT_QC_MENUS = {
    "biz:qcWorkOrder:list",
    "biz:myQcTask:list",
    "production:work-order:list",
    "production:my-task:list",
    "cell-storage:sample-info:list",
}


def _session():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(
        engine,
        tables=[
            SysMenu.__table__,
            SysPost.__table__,
            SysPostRole.__table__,
            SysRole.__table__,
            SysRoleMenu.__table__,
            SysUser.__table__,
            SysUserPost.__table__,
            SysUserRole.__table__,
        ],
    )
    return sessionmaker(bind=engine)()


def _thread_session():
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(
        engine,
        tables=[
            SysMenu.__table__,
            SysPost.__table__,
            SysPostRole.__table__,
            SysRole.__table__,
            SysRoleMenu.__table__,
            SysUser.__table__,
            SysUserPost.__table__,
            SysUserRole.__table__,
        ],
    )
    return sessionmaker(bind=engine)()


def _add_user(db, *, user_id="u1", tenant_no="tenant-a"):
    db.add(
        SysUser(
            id=user_id,
            user_no=user_id,
            nick_name=user_id,
            password="unused",
            status="1",
            tenant_no=tenant_no,
            create_dept_id="",
            del_flag=0,
        )
    )


def _add_role(db, *, role_id, tenant_no, role_name="普通角色"):
    db.add(
        SysRole(
            id=role_id,
            role_name=role_name,
            status="1",
            tenant_no=tenant_no,
            create_dept_id="",
            del_flag=0,
        )
    )


def _add_menu_grant(db, *, role_id, tenant_no, code, suffix):
    menu_id = f"menu-{suffix}"
    db.add_all(
        [
            SysMenu(
                id=menu_id,
                menu_name=code,
                parent_id="root",
                menu_type="C",
                path=f"/{suffix}",
                perms=code,
                status="1",
                tenant_no=tenant_no,
                create_dept_id="",
                del_flag=0,
            ),
            SysRoleMenu(
                id=f"rm-{role_id}-{suffix}",
                role_id=role_id,
                menu_id=menu_id,
                tenant_no=tenant_no,
                create_dept_id="",
                del_flag=0,
            ),
        ]
    )


def _bind_direct(db, *, user_id, role_id, tenant_no, suffix):
    db.add(
        SysUserRole(
            id=f"ur-{suffix}",
            user_id=user_id,
            role_id=role_id,
            tenant_no=tenant_no,
            create_dept_id="",
            del_flag=0,
        )
    )


def _bind_post(db, *, user_id, role_id, tenant_no, suffix):
    post_id = f"post-{suffix}"
    db.add_all(
        [
            SysPost(
                id=post_id,
                post_name="授权职位",
                post_code=post_id,
                status="1",
                tenant_no=tenant_no,
                create_dept_id="",
                del_flag=0,
            ),
            SysUserPost(
                id=f"up-{suffix}",
                user_id=user_id,
                post_id=post_id,
                tenant_no=tenant_no,
                create_dept_id="",
                del_flag=0,
            ),
            SysPostRole(
                id=f"pr-{suffix}",
                post_id=post_id,
                role_id=role_id,
                tenant_no=tenant_no,
                create_dept_id="",
                del_flag=0,
            ),
        ]
    )


def _strict_checker():
    checker = getattr(menu_guard, "strict_user_has_menu_perm", None)
    assert callable(checker), "menu_guard must expose strict_user_has_menu_perm"
    return checker


def _strict_dependency(code):
    factory = getattr(menu_guard, "require_strict_menu_code", None)
    assert callable(factory), "menu_guard must expose require_strict_menu_code"
    return factory(code)


def test_strict_permission_uses_direct_and_post_role_union():
    db = _session()
    _add_user(db)
    _add_role(db, role_id="market", tenant_no="tenant-a", role_name="市场专员")
    _add_role(db, role_id="finance", tenant_no="tenant-a", role_name="财务人员")
    _add_menu_grant(
        db,
        role_id="market",
        tenant_no="tenant-a",
        code="mini:market:specialist:payment",
        suffix="mini",
    )
    _add_menu_grant(
        db,
        role_id="finance",
        tenant_no="tenant-a",
        code=PAYMENT_CONFIRM,
        suffix="finance",
    )
    _bind_direct(
        db, user_id="u1", role_id="market", tenant_no="tenant-a", suffix="market"
    )
    _bind_post(
        db, user_id="u1", role_id="finance", tenant_no="tenant-a", suffix="finance"
    )
    db.commit()

    checker = _strict_checker()
    user = {"user_id": "u1", "tenant_no": "forged", "role": "admin"}
    assert checker(db, user, PAYMENT_CONFIRM)
    assert not checker(db, user, REFUND_CONFIRM)
    assert asyncio.run(_strict_dependency(PAYMENT_CONFIRM)(user=user, db=db)) is user


def test_strict_permission_ignores_cross_tenant_grants():
    db = _session()
    _add_user(db, tenant_no="tenant-a")
    _add_role(db, role_id="market", tenant_no="tenant-a", role_name="市场专员")
    _bind_direct(
        db, user_id="u1", role_id="market", tenant_no="tenant-a", suffix="market"
    )
    _add_role(db, role_id="finance-b", tenant_no="tenant-b", role_name="财务人员")
    _add_menu_grant(
        db,
        role_id="finance-b",
        tenant_no="tenant-b",
        code=PAYMENT_CONFIRM,
        suffix="finance-b",
    )
    _bind_direct(
        db,
        user_id="u1",
        role_id="finance-b",
        tenant_no="tenant-b",
        suffix="finance-b",
    )
    db.commit()

    assert not _strict_checker()(
        db,
        {"user_id": "u1", "tenant_no": "tenant-b", "role": "super_admin"},
        PAYMENT_CONFIRM,
    )


@pytest.mark.parametrize("role_name", ["超级市场专员", "超级管理员"])
def test_strict_permission_rejects_privileged_role_name_without_menu(role_name):
    db = _session()
    _add_user(db)
    _add_role(
        db,
        role_id="super",
        tenant_no="tenant-a",
        role_name=role_name,
    )
    _bind_direct(
        db, user_id="u1", role_id="super", tenant_no="tenant-a", suffix="super"
    )
    db.commit()

    user = {"user_id": "u1", "tenant_no": "forged", "role": "admin"}
    assert not _strict_checker()(db, user, REFUND_CONFIRM)
    with pytest.raises(HTTPException) as exc_info:
        asyncio.run(_strict_dependency(REFUND_CONFIRM)(user=user, db=db))
    assert exc_info.value.status_code == 403


def test_strict_permission_allows_super_admin_only_with_explicit_menu_grant():
    db = _session()
    _add_user(db)
    _add_role(db, role_id="super", tenant_no="tenant-a", role_name="超级管理员")
    _add_menu_grant(
        db,
        role_id="super",
        tenant_no="tenant-a",
        code=REFUND_CONFIRM,
        suffix="refund",
    )
    _bind_direct(
        db, user_id="u1", role_id="super", tenant_no="tenant-a", suffix="super"
    )
    db.commit()

    user = {"user_id": "u1", "tenant_no": "forged", "role": "super_admin"}
    assert _strict_checker()(db, user, REFUND_CONFIRM)
    assert asyncio.run(_strict_dependency(REFUND_CONFIRM)(user=user, db=db)) is user


def test_strict_dependency_rejects_market_only_role():
    db = _session()
    _add_user(db)
    _add_role(db, role_id="market", tenant_no="tenant-a", role_name="地区经理")
    _add_menu_grant(
        db,
        role_id="market",
        tenant_no="tenant-a",
        code="mini:market:manager:protocol",
        suffix="mini",
    )
    _bind_direct(
        db, user_id="u1", role_id="market", tenant_no="tenant-a", suffix="market"
    )
    db.commit()

    with pytest.raises(HTTPException) as exc_info:
        asyncio.run(
            _strict_dependency(PROTOCOL_AUDIT)(
                user={"user_id": "u1", "tenant_no": "tenant-a", "role": "admin"},
                db=db,
            )
        )

    assert exc_info.value.status_code == 403


def _route_strict_codes(router, endpoint_name):
    route = next(r for r in router.routes if r.endpoint.__name__ == endpoint_name)
    codes = set()
    for dependency in route.dependant.dependencies:
        codes.update(getattr(dependency.call, "strict_menu_codes", ()))
    return codes


@pytest.mark.parametrize(
    ("module_name", "endpoint_names", "expected_codes"),
    [
        (
            "research_donation",
            [
                "create_research_donation",
                "update_research_donation",
            ],
            set(),
        ),
        (
            "protocol_sample_payment",
            ["confirm_payment_transfer", "reject_payment_transfer"],
            set(),
        ),
        (
            "presign_order",
            ["confirm_presign_transfer", "reject_presign_transfer"],
            set(),
        ),
        (
            "sample_cert_insurance",
            ["put_sample_storage_cert", "put_sample_insurance"],
            set(),
        ),
        (
            "sample_storage_cert_qc",
            ["put_sample_storage_cert_qc"],
            STORAGE_CERT_QC_MENUS,
        ),
    ],
)
def test_sensitive_write_routes_declare_strict_menu_dependencies(
    module_name, endpoint_names, expected_codes
):
    module = __import__(f"app.api.{module_name}", fromlist=["router"])
    for endpoint_name in endpoint_names:
        assert _route_strict_codes(module.router, endpoint_name) == expected_codes


def test_research_donation_audit_route_has_no_strict_menu_dependency():
    """科研捐赠审核只保留运营端登录，不再要求审核菜单权限。"""
    from app.api import research_donation

    assert _route_strict_codes(research_donation.router, "submit_donation_audit") == set()


def test_collection_protocol_write_routes_have_no_strict_menu_dependencies():
    """采集协议单接口仅保留运营端登录守卫，不再挂菜单权限码。"""
    from app.api import collection_protocol

    write_endpoints = [
        "create_collection_protocol",
        "update_collection_protocol",
        "delete_collection_protocol",
        "submit_collection_protocol_audit",
        "post_receipt_confirm",
        "post_receipt_reject",
        "post_protocol_invoice_request",
        "complete_protocol_invoice_request",
        "post_submit_refund_finance",
        "post_submit_refund_finance_batch",
        "post_refund_confirm",
        "post_refund_reject",
        "update_fullmoon_todo_insurance",
        "update_storage_cert_handle",
        "update_fullmoon_todo_handle",
    ]
    for endpoint_name in write_endpoints:
        assert _route_strict_codes(collection_protocol.router, endpoint_name) == set()


def test_protocol_refund_record_routes_have_no_strict_menu_dependencies():
    """采集协议单上的退费登记只保留运营端登录，不再要求退费确认菜单。"""
    from app.api import protocol_refund_record

    for endpoint_name in (
        "list_protocol_refund_records",
        "create_protocol_refund_record",
        "update_protocol_refund_record",
        "delete_protocol_refund_record",
    ):
        assert _route_strict_codes(protocol_refund_record.router, endpoint_name) == set()


def test_collection_protocol_write_paths_keep_payment_detail_guard_only():
    source = Path("app/api/collection_protocol.py").read_text()
    assert "assert_protocol_sensitive_fields" in source
    assert "require_strict_menu_code" not in source
    assert "assert_strict_menu_code" not in source


def test_research_donation_write_paths_assert_sensitive_field_permissions():
    source = Path("app/api/research_donation.py").read_text()
    assert "assert_donation_sensitive_fields" in source
    assert "require_strict_menu_code" not in source
    assert "assert_strict_menu_code" not in source


def _permission_user(db, *codes):
    _add_user(db)
    _add_role(db, role_id="protocol", tenant_no="tenant-a", role_name="协议录入")
    _add_menu_grant(
        db,
        role_id="protocol",
        tenant_no="tenant-a",
        code=PROTOCOL_LIST,
        suffix="protocol",
    )
    _bind_direct(
        db,
        user_id="u1",
        role_id="protocol",
        tenant_no="tenant-a",
        suffix="protocol",
    )
    for index, code in enumerate(codes):
        role_id = f"extra-{index}"
        _add_role(db, role_id=role_id, tenant_no="tenant-a")
        _add_menu_grant(
            db,
            role_id=role_id,
            tenant_no="tenant-a",
            code=code,
            suffix=role_id,
        )
        _bind_direct(
            db,
            user_id="u1",
            role_id=role_id,
            tenant_no="tenant-a",
            suffix=role_id,
        )
    db.commit()
    return {"user_id": "u1", "tenant_no": "tenant-a", "role": "admin"}


def test_protocol_sensitive_fields_no_longer_require_menu_permissions():
    from app.api.collection_protocol import (
        ProtocolCreateBody,
        ProtocolUpdateBody,
        assert_protocol_sensitive_fields,
    )

    db = _session()
    user = _permission_user(db)
    assert_protocol_sensitive_fields(db, user, ProtocolCreateBody(), creating=True)
    assert_protocol_sensitive_fields(
        db,
        user,
        ProtocolCreateBody(protocolStatus="sxz"),
        creating=True,
    )
    assert_protocol_sensitive_fields(
        db,
        user,
        ProtocolUpdateBody(storageCert={}),
        creating=False,
    )


def test_donation_sensitive_fields_no_longer_require_menu_permissions():
    from app.api.research_donation import (
        DonationCreateBody,
        DonationUpdateBody,
        assert_donation_sensitive_fields,
    )

    db = _session()
    user = _permission_user(db)
    assert_donation_sensitive_fields(db, user, DonationCreateBody(), creating=True)
    for body, creating in (
        (DonationCreateBody(donationStatus="sxz"), True),
        (DonationCreateBody(sourceType="cell_storage"), True),
        (DonationUpdateBody(sourceProtocolId="p1"), False),
    ):
        assert_donation_sensitive_fields(db, user, body, creating=creating)


def test_market_only_user_is_stopped_before_storage_cert_qc_write(monkeypatch):
    """储存证质控写接口仍挂严格菜单码：无对应菜单时应 403。"""
    from app.api import sample_storage_cert_qc

    db = _thread_session()
    _add_user(db)
    _add_role(db, role_id="market", tenant_no="tenant-a", role_name="市场专员")
    _add_menu_grant(
        db,
        role_id="market",
        tenant_no="tenant-a",
        code="mini:market:specialist:protocol",
        suffix="mini-protocol",
    )
    _bind_direct(
        db, user_id="u1", role_id="market", tenant_no="tenant-a", suffix="market"
    )
    db.commit()

    called = []

    def forbidden_call(*args, **kwargs):
        called.append((args, kwargs))
        raise AssertionError("sensitive CRUD must not be called")

    monkeypatch.setattr(sample_storage_cert_qc.svc, "upsert", forbidden_call)

    app = FastAPI()
    app.include_router(sample_storage_cert_qc.router)

    user = {"user_id": "u1", "tenant_no": "tenant-a", "role": "admin"}

    def override_db():
        yield db

    app.dependency_overrides[get_current_user] = lambda: user
    app.dependency_overrides[get_db] = override_db

    client = TestClient(app)
    response = client.request(
        "put", "/web/admin/sample-storage-cert-qc/S1", json={"payload": {}}
    )
    assert response.status_code == 403, response.text
    assert called == []


def test_quick_register_writes_skip_menu_and_data_scope():
    """客户服务七类登记不再校验菜单码，也不按数据权限拦截样本。"""
    api_source = Path("app/api/protocol_sample_collection.py").read_text()
    crud_source = Path("app/crud/sample_info_quick_register.py").read_text()
    assert "assert_strict_menu_code" not in api_source
    assert "apply_data_scope" not in crud_source


@pytest.mark.parametrize("kind", ["storage_cert", "insurance", "all"])
def test_quick_register_does_not_require_storage_cert_menu(monkeypatch, kind):
    """客户服务管理登记储存证/保险不再要求储存证菜单权限。"""
    from app.api import protocol_sample_collection

    db = _session()
    user = _permission_user(db)
    called = []

    def allowed_call(*args, **kwargs):
        called.append((args, kwargs))
        return {"success": 1, "fail": 0}

    monkeypatch.setattr(
        protocol_sample_collection.quick_svc,
        "batch_handle_storage_certs",
        allowed_call,
    )
    monkeypatch.setattr(
        protocol_sample_collection.quick_svc,
        "batch_handle_insurance",
        allowed_call,
    )
    body = protocol_sample_collection.QuickRegisterBody(
        kind=kind,
        receptionSource="cell_storage",
        items=[{"sampleCode": "S1", "protocolId": "P1"}],
        handleTime="2026-10-14 10:00:00",
    )

    result = asyncio.run(
        protocol_sample_collection.quick_register_sample_info(
            body=body,
            db=db,
            user=user,
        )
    )

    assert result["code"] == 200
    assert called


@pytest.mark.parametrize(
    "payment_detail",
    [
        {"installmentNo": 1, "payStatus": "yjf"},
        {"installmentNo": 1, "PayStatus": "yjf"},
        {"installmentNo": 1, "financeConfirmAttachments": "secret.pdf"},
        {
            "installmentNo": 1,
            "financeConfirmAttachmentFiles": [{"url": "secret.pdf"}],
        },
        {"installmentNo": 1, "financeRemark": "confirmed"},
        {"installmentNo": 1, "refundAmount": 100},
        {"installmentNo": 1, "refundRemark": "refund"},
        {"installmentNo": 1, "refundAttachments": "refund.pdf"},
        {
            "installmentNo": 1,
            "refundAttachmentFiles": [{"url": "refund.pdf"}],
        },
        {
            "installmentNo": 1,
            "refundFinanceConfirmAttachments": "finance.pdf",
        },
        {
            "installmentNo": 1,
            "refundFinanceConfirmAttachmentFiles": [{"url": "finance.pdf"}],
        },
        {"installmentNo": 1, "refundFinanceRemark": "confirmed"},
    ],
)
def test_protocol_create_rejects_nested_finance_and_refund_fields_before_crud(
    monkeypatch, payment_detail
):
    from app.api import collection_protocol

    called = []

    def forbidden_call(*args, **kwargs):
        called.append((args, kwargs))
        raise AssertionError("protocol CRUD must not be called")

    monkeypatch.setattr(
        collection_protocol.cp_svc,
        "create_collection_protocol",
        forbidden_call,
    )
    body = collection_protocol.ProtocolCreateBody(
        paymentModelVersion=1,
        paymentDetails=[payment_detail],
    )

    with pytest.raises(HTTPException) as exc_info:
        asyncio.run(
            collection_protocol.create_collection_protocol(
                body=body,
                user={"user_id": "u1", "role": "admin"},
                db=None,
            )
        )

    assert exc_info.value.status_code == 400
    assert called == []


def test_protocol_update_rejects_snake_case_nested_sensitive_fields_before_crud(
    monkeypatch,
):
    from app.api import collection_protocol

    called = []

    def forbidden_call(*args, **kwargs):
        called.append((args, kwargs))
        raise AssertionError("protocol CRUD must not be called")

    monkeypatch.setattr(
        collection_protocol.cp_svc,
        "update_collection_protocol",
        forbidden_call,
    )
    body = collection_protocol.ProtocolUpdateBody(
        payment_details=[
            {
                "installment_no": 1,
                "finance_confirm_attachment_files": [{"url": "secret.pdf"}],
            }
        ]
    )

    with pytest.raises(HTTPException) as exc_info:
        asyncio.run(
            collection_protocol.update_collection_protocol(
                protocol_id="p1",
                body=body,
                user={"user_id": "u1", "role": "admin"},
                db=None,
            )
        )

    assert exc_info.value.status_code == 400
    assert called == []


def test_payment_detail_sensitive_field_inventory_matches_crud_write_fields():
    from app.api.collection_protocol import _PAYMENT_DETAIL_SENSITIVE_FIELDS

    assert _PAYMENT_DETAIL_SENSITIVE_FIELDS == {
        "pay_status",
        "payStatus",
        "finance_confirm_attachments",
        "financeConfirmAttachments",
        "finance_confirm_attachment_files",
        "financeConfirmAttachmentFiles",
        "finance_remark",
        "financeRemark",
        "refund_amount",
        "refundAmount",
        "refund_remark",
        "refundRemark",
        "refund_attachments",
        "refundAttachments",
        "refund_attachment_files",
        "refundAttachmentFiles",
        "refund_finance_confirm_attachments",
        "refundFinanceConfirmAttachments",
        "refund_finance_confirm_attachment_files",
        "refundFinanceConfirmAttachmentFiles",
        "refund_finance_remark",
        "refundFinanceRemark",
    }