zkxw/backend/tests/test_web_presign_permissions.py

"""预签约 Web 入口使用真实角色范围查询，在越权时不得调用任何读写服务。"""
import asyncio
from datetime import datetime
from unittest.mock import Mock

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.api import presign_order as api
from app.models.base import Base
from app.models.models import (
    BizCollectionProtocol, BizPresignOrder, BizPresignTransfer, SysRole, SysRoleDataScopeDetail,
    SysUser, SysUserRole,
)

USER = {'user_id': 'u1', 'role': 'admin', 'tenant_no': 'tenant-a', 'dept_id': 'dept-a'}


@pytest.fixture
def db():
    engine = create_engine('sqlite:///:memory:')
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine)()
    common = dict(tenant_no='tenant-a', create_dept_id='dept-a', create_time=datetime.now(), del_flag=0)
    session.add_all([
        SysUser(id='u1', user_no='u1', nick_name='用户', password='', dept_id='dept-a', **common),
        SysRole(id='self', role_name='普通人员', data_scope='4', status='1', **common),
        SysUserRole(id='ur', user_id='u1', role_id='self', **common),
    ])
    for owner in ('u1', 'u2'):
        session.add_all([
            BizCollectionProtocol(id='cp-' + owner, protocol_no='CP-' + owner, create_by=owner, **common),
            BizPresignOrder(id='p-' + owner, protocol_no='P-' + owner, order_status='active', create_by=owner, **common),
            BizPresignTransfer(id='t-' + owner, presign_order_id='p-' + owner, pay_status='jfqrz', transfer_amount=10, create_by=owner, **common),
        ])
    session.commit()
    yield session
    session.close()
    engine.dispose()


@pytest.mark.parametrize('action', [
    'get', 'update', 'delete', 'convert', 'create_transfer',
    'update_transfer', 'submit', 'withdraw', 'confirm', 'reject',
])
@pytest.mark.parametrize('access', ['denied', 'self', 'union'])
def test_presign_endpoint_enforces_union_data_scope(db, monkeypatch, action, access):
    if access == 'union':
        common = dict(tenant_no='tenant-a', create_dept_id='', del_flag=0)
        db.add_all([
            SysRole(id='custom', role_name='自定义', data_scope='5', status='1', **common),
            SysUserRole(id='ur2', user_id='u1', role_id='custom', **common),
            SysRoleDataScopeDetail(id='detail', role_id='custom', scope_type='user', scope_value='u2', **common),
        ])
        db.commit()
    owner = 'u1' if access == 'self' else 'u2'
    cases = {
        'get': (api.get_presign_order, 'get_presign_order', {'order_id': 'p-' + owner}),
        'update': (api.update_presign_order, 'update_presign_order', {'order_id': 'p-' + owner, 'body': api.PresignOrderBody(remark='test')}),
        'delete': (api.delete_presign_order, 'soft_delete_presign_order', {'order_id': 'p-' + owner}),
        'convert': (api.convert_presign_order, 'convert_to_protocol', {'order_id': 'p-' + owner}),
        'create_transfer': (api.create_presign_transfer, 'add_transfer', {'order_id': 'p-' + owner, 'body': api.PresignTransferBody(transferAmount=10)}),
        'update_transfer': (api.update_presign_transfer, 'update_transfer_draft', {'transfer_id': 't-' + owner, 'body': api.PresignTransferBody(transferAmount=10)}),
        'submit': (api.submit_presign_transfer_finance, 'submit_receipt_finance', {'transfer_id': 't-' + owner}),
        'withdraw': (api.withdraw_presign_transfer, 'withdraw_receipt', {'transfer_id': 't-' + owner}),
        'confirm': (api.confirm_presign_transfer, 'confirm_receipt', {'transfer_id': 't-' + owner, 'body': None}),
        'reject': (api.reject_presign_transfer, 'reject_receipt', {'transfer_id': 't-' + owner}),
    }
    endpoint, service_name, kwargs = cases[action]
    service = Mock(return_value={'id': 'result'})
    monkeypatch.setattr(api.po_svc, service_name, service)
    monkeypatch.setattr(api.po_svc, 'order_to_dict', lambda *args, **kwargs: {'id': 'result'})
    monkeypatch.setattr(api.po_svc, '_transfer_to_dict', lambda *args, **kwargs: {'id': 'result'})
    result = asyncio.run(endpoint(db=db, user=USER, **kwargs))
    if access == 'denied':
        assert result['code'] == 403
        service.assert_not_called()
    else:
        assert result['code'] == 200
        service.assert_called_once()


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

    audit = Mock()
    monkeypatch.setattr(collection_protocol.cp_svc, 'create_audit', audit)
    result = asyncio.run(collection_protocol.submit_collection_protocol_audit(
        protocol_id='cp-u2',
        body=collection_protocol.AuditCreateBody(auditStatus='approved'),
        db=db, user=USER,
    ))
    assert result['code'] == 403
    audit.assert_not_called()


@pytest.mark.parametrize('action', ['create', 'update', 'transfer', 'convert'])
def test_web_presign_writes_use_server_department(db, monkeypatch, action):
    cases = {
        'create': (api.create_presign_order, 'create_presign_order', {'body': api.PresignOrderBody(motherName='客户')}),
        'update': (api.update_presign_order, 'update_presign_order', {'order_id': 'p-u1', 'body': api.PresignOrderBody(remark='test')}),
        'transfer': (api.create_presign_transfer, 'add_transfer', {'order_id': 'p-u1', 'body': api.PresignTransferBody(transferAmount=10)}),
        'convert': (api.convert_presign_order, 'convert_to_protocol', {'order_id': 'p-u1'}),
    }
    endpoint, service_name, kwargs = cases[action]
    service = Mock(return_value={'id': 'result'})
    monkeypatch.setattr(api.po_svc, service_name, service)
    monkeypatch.setattr(api.po_svc, 'order_to_dict', lambda *args, **kwargs: {'id': 'result'})
    monkeypatch.setattr(api.po_svc, '_transfer_to_dict', lambda *args, **kwargs: {'id': 'result'})
    result = asyncio.run(endpoint(db=db, user={**USER, 'dept_id': 'untrusted-dept'}, **kwargs))
    assert result['code'] == 200
    assert service.call_args.kwargs['create_dept_id'] == 'dept-a'


def test_collection_protocol_create_uses_server_department(db, monkeypatch):
    from types import SimpleNamespace
    from app.api import collection_protocol

    create = Mock(return_value=SimpleNamespace(id='new-protocol', payment_model_version=2))
    monkeypatch.setattr(collection_protocol.cp_svc, 'create_collection_protocol', create)
    monkeypatch.setattr(collection_protocol.cp_svc, '_protocol_to_dict', lambda obj: {'id': obj.id})
    monkeypatch.setattr(collection_protocol, '_persist_protocol_sub_resources', Mock())
    result = asyncio.run(collection_protocol.create_collection_protocol(
        body=collection_protocol.ProtocolCreateBody(), db=db,
        user={**USER, 'dept_id': 'untrusted-dept'},
    ))
    assert result['code'] == 200
    assert create.call_args.kwargs['create_dept_id'] == 'dept-a'