46 lines
2.6 KiB
Python
46 lines
2.6 KiB
Python
"""两类缴费的财务确认/退回成功路径与重复操作拒绝，纯内存数据库。"""
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.crud import presign_order, protocol_sample_payment
|
|
from app.models.base import Base
|
|
from app.models.models import (
|
|
BizCollectionProtocol, BizPresignOrder, BizPresignTransfer,
|
|
BizProtocolSamplePayment, BizProtocolPaymentSchedule, BizProtocolPaymentTransfer,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['presign', 'sample'])
|
|
@pytest.mark.parametrize('operation,expected', [('confirm', 'yjf'), ('reject', 'dfj')])
|
|
def test_finance_transition_succeeds_once_and_rejects_duplicate(kind, operation, expected):
|
|
engine = create_engine('sqlite:///:memory:')
|
|
Base.metadata.create_all(engine)
|
|
with sessionmaker(bind=engine)() as db:
|
|
common = dict(tenant_no='tenant-a', create_dept_id='', del_flag=0)
|
|
if kind == 'presign':
|
|
db.add(BizPresignOrder(id='p', order_status='active', protocol_no='P', **common))
|
|
db.add(BizPresignTransfer(id='t', presign_order_id='p', pay_status='jfqrz', transfer_amount=100, **common))
|
|
crud, model = presign_order, BizPresignTransfer
|
|
else:
|
|
db.add_all([
|
|
BizCollectionProtocol(id='p', protocol_no='P', **common),
|
|
BizProtocolSamplePayment(id='pay', collection_protocol_id='p', sample_code='S', pay_mode='dj', contract_amount=100, **common),
|
|
BizProtocolPaymentSchedule(id='sch', sample_payment_id='pay', collection_protocol_id='p', sample_code='S', schedule_no=1, due_amount=100, schedule_status='open', **common),
|
|
BizProtocolPaymentTransfer(id='t', schedule_id='sch', sample_payment_id='pay', collection_protocol_id='p', sample_code='S', transfer_amount=100, pay_status='jfqrz', **common),
|
|
])
|
|
crud, model = protocol_sample_payment, BizProtocolPaymentTransfer
|
|
db.commit()
|
|
function = crud.confirm_receipt if operation == 'confirm' else crud.reject_receipt
|
|
result = function(db, transfer_id='t', tenant_no='tenant-a', user_id='finance')
|
|
assert result.pay_status == expected
|
|
with pytest.raises(ValueError):
|
|
function(db, transfer_id='t', tenant_no='tenant-a', user_id='finance')
|
|
assert db.query(model).filter_by(id='t').one().pay_status == expected
|
|
if kind == 'sample' and operation == 'confirm':
|
|
assert (
|
|
db.query(BizProtocolPaymentSchedule).filter_by(id='sch').one().schedule_status
|
|
== 'cleared'
|
|
)
|
|
engine.dispose()
|