130 lines
6.7 KiB
Python
130 lines
6.7 KiB
Python
"""预签约 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'
|