zkxw/backend/tests/test_payment_finance_status_regression.py

"""两类缴费的财务确认/退回成功路径与重复操作拒绝，纯内存数据库。"""
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()