455 lines
20 KiB
Python
455 lines
20 KiB
Python
"""BX/充值通知快照：只取当前节点和真实审批结果，不把发放当审批。"""
|
|
from datetime import date, datetime, timedelta
|
|
import importlib
|
|
import importlib.util
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.models.bx_models import (
|
|
BxAdvanceApply, BxAdvanceApproval, BxReimburseApply, BxReimburseApproval,
|
|
BxWorkflowDef, BxWorkflowNode,
|
|
)
|
|
from app.models.pl_models import PlVehicleRechargeApply, PlVehicleRechargeApproval
|
|
|
|
|
|
@pytest.fixture
|
|
def db():
|
|
engine = create_engine('sqlite://')
|
|
for model in (BxAdvanceApply, BxAdvanceApproval, BxReimburseApply,
|
|
BxReimburseApproval, BxWorkflowDef, BxWorkflowNode,
|
|
PlVehicleRechargeApply, PlVehicleRechargeApproval):
|
|
model.__table__.create(engine)
|
|
session = sessionmaker(bind=engine, autoflush=False)()
|
|
yield session
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
def snapshot(db, kind):
|
|
name = 'app.services.wechat_notice_bx_sources'
|
|
assert importlib.util.find_spec(name) is not None, 'BX/充值审批通知快照适配器尚未实现'
|
|
return importlib.import_module(name).get_snapshot(db, kind, 'apply', 't1')
|
|
|
|
|
|
def seed(db, kind, status=None, disburse_first=False):
|
|
recharge = kind == 'pl_recharge'
|
|
db.add(BxWorkflowDef(id='wf', biz_type='vehicle_recharge' if recharge else kind,
|
|
wf_name='审批流程', tenant_no='t1'))
|
|
for i in (1, 2):
|
|
db.add(BxWorkflowNode(id=f'n{i}', wf_id='wf', node_name=f'审批{i}', node_order=i,
|
|
approver_id=f'u{i}', tenant_no='t1',
|
|
action_type='disburse' if recharge and (i == 2 or disburse_first) else 'approve'))
|
|
fields = dict(id='apply', applicant_id='applicant', create_by='operator', tenant_no='t1',
|
|
wf_id='wf', status=status or ('in_approval' if recharge else 'pending'),
|
|
current_node='n1' if recharge else '审批1')
|
|
if recharge:
|
|
row = PlVehicleRechargeApply(**fields, apply_no='PL001', license_plate='鄂A00001',
|
|
recharge_type='fuel', apply_date=date.today(), approval_round=2)
|
|
else:
|
|
fields.update(apply_code='BX001', applicant_name='申请人', amount=100)
|
|
row = (BxAdvanceApply(**fields, reason='备用金') if kind == 'bx_advance' else
|
|
BxReimburseApply(**fields, title='报销', reimburse_date=date.today(),
|
|
expense_id='expense', expense_name='差旅'))
|
|
db.add(row)
|
|
db.flush()
|
|
return row
|
|
|
|
|
|
def approval(db, kind, outcome='approved', uid='u2', when=None, round_no=2):
|
|
model = {'bx_advance': BxAdvanceApproval, 'bx_reimburse': BxReimburseApproval,
|
|
'pl_recharge': PlVehicleRechargeApproval}[kind]
|
|
fields = dict(id=f'{outcome}-{uid}-{round_no}', apply_id='apply', tenant_no='t1',
|
|
approver_id=uid, approver_name=uid, node_name='审批2',
|
|
approval_status=outcome, approval_time=when or datetime(2026, 10, 27, 12))
|
|
fields.update(dict(approval_round=round_no, node_id='n2') if kind == 'pl_recharge' else dict(node_order=2))
|
|
record = model(**fields)
|
|
db.add(record)
|
|
db.flush()
|
|
return record
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse', 'pl_recharge'])
|
|
def test_only_current_approver_and_real_applicant(db, kind):
|
|
row = seed(db, kind)
|
|
result = snapshot(db, kind)
|
|
assert [(p.task_key, p.user_id) for p in result.pending] == [('n1', 'u1')]
|
|
assert result.applicant_id == 'applicant'
|
|
assert result.result is None
|
|
assert result.approver_id == ''
|
|
assert result.page == ''
|
|
row.current_node = 'n2' if kind == 'pl_recharge' else '审批2'
|
|
db.flush()
|
|
result = snapshot(db, kind)
|
|
assert [p.user_id for p in result.pending] == ([] if kind == 'pl_recharge' else ['u2'])
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse', 'pl_recharge'])
|
|
@pytest.mark.parametrize('state', ['draft', 'cancelled'])
|
|
def test_inactive_row_suppresses_stale_node_and_approval(db, kind, state):
|
|
seed(db, kind, status=state)
|
|
approval(db, kind)
|
|
result = snapshot(db, kind)
|
|
assert result.pending == []
|
|
assert result.result is None
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse', 'pl_recharge'])
|
|
def test_result_uses_latest_approval_time_not_node_order_or_operator(db, kind):
|
|
row = seed(db, kind, status='rejected')
|
|
approval(db, kind, when=datetime(2026, 10, 27, 11))
|
|
last = approval(db, kind, outcome='rejected', uid='u1')
|
|
result = snapshot(db, kind)
|
|
assert result.result == 'rejected'
|
|
assert result.pending == []
|
|
assert result.approver_id == 'u1'
|
|
assert result.occurred_at == last.approval_time
|
|
assert result.page == {
|
|
'bx_advance': 'pages/bx/advance/detail?id=apply',
|
|
'bx_reimburse': 'pages/bx/reimburse/detail?id=apply',
|
|
'pl_recharge': 'pages/pl/vehicle-recharge-apply/form?id=apply',
|
|
}[kind]
|
|
row.status = 'in_approval' if kind == 'pl_recharge' else 'pending'
|
|
db.flush()
|
|
assert snapshot(db, kind).result is None
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse'])
|
|
def test_bx_approved_requires_real_record(db, kind):
|
|
seed(db, kind, status='approved')
|
|
assert snapshot(db, kind).result is None
|
|
record = approval(db, kind)
|
|
result = snapshot(db, kind)
|
|
assert result.result == 'approved'
|
|
assert result.occurred_at == record.approval_time
|
|
|
|
|
|
def test_recharge_approval_completes_before_disbursement(db):
|
|
row = seed(db, 'pl_recharge')
|
|
row.current_node = 'n2'
|
|
db.flush()
|
|
approval(db, 'pl_recharge', round_no=1)
|
|
assert snapshot(db, 'pl_recharge').result is None
|
|
approved = approval(db, 'pl_recharge')
|
|
result = snapshot(db, 'pl_recharge')
|
|
assert result.result == 'approved'
|
|
assert result.pending == []
|
|
assert result.approver_id == 'u2'
|
|
assert result.occurred_at == approved.approval_time
|
|
row.status = 'disbursed'
|
|
approval(db, 'pl_recharge', outcome='disbursed', uid='finance',
|
|
when=approved.approval_time + timedelta(hours=1))
|
|
result = snapshot(db, 'pl_recharge')
|
|
assert result.result == 'approved'
|
|
assert result.approver_id == 'u2'
|
|
assert result.occurred_at == approved.approval_time
|
|
|
|
|
|
def test_recharge_direct_disbursement_does_not_invent_approval(db):
|
|
row = seed(db, 'pl_recharge', disburse_first=True)
|
|
assert snapshot(db, 'pl_recharge').pending == []
|
|
assert snapshot(db, 'pl_recharge').result is None
|
|
row.status = 'disbursed'
|
|
approval(db, 'pl_recharge', outcome='disbursed', uid='finance')
|
|
assert snapshot(db, 'pl_recharge').result is None
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse', 'pl_recharge'])
|
|
def test_tenant_and_soft_deleted_application_are_excluded(db, kind):
|
|
row = seed(db, kind)
|
|
row.tenant_no = 'other'
|
|
db.flush()
|
|
assert snapshot(db, kind) is None
|
|
row.tenant_no = 't1'
|
|
row.del_flag = 1
|
|
db.flush()
|
|
assert snapshot(db, kind) is None
|
|
|
|
|
|
@pytest.fixture
|
|
def live(db, monkeypatch):
|
|
from app.core.bx_auth import BxUserContext
|
|
from app.models.bx_models import BxInvoiceLedger, BxInvoiceWalletSummary
|
|
from app.models.models import SysUser
|
|
from app.models.wechat_notice import WechatApprovalCycle, WechatNotice
|
|
from app.services import wechat_approval_notices as notices
|
|
from app.services.bx import base
|
|
from app.services import pl_vehicle_recharge_service as recharge
|
|
from app.services import pl_vehicle_recharge_approval_service as recharge_approval
|
|
for model in (SysUser, BxInvoiceLedger, BxInvoiceWalletSummary, WechatApprovalCycle, WechatNotice):
|
|
model.__table__.create(db.get_bind())
|
|
for uid in ('applicant', 'operator', 'u1', 'u2'):
|
|
db.add(SysUser(id=uid, tenant_no='t1', user_no=uid, nick_name=uid))
|
|
db.commit()
|
|
monkeypatch.setattr(notices.settings, 'WECHAT_OA_ENABLED', True)
|
|
monkeypatch.setattr(notices.settings, 'WECHAT_OA_TEMPLATE_PENDING', 'pending-template')
|
|
monkeypatch.setattr(notices.settings, 'WECHAT_OA_TEMPLATE_RESULT', 'result-template')
|
|
monkeypatch.setattr(base, 'assert_bx_row_in_scope', lambda *a, **kw: None)
|
|
monkeypatch.setattr(recharge, 'assert_pl_row_in_scope', lambda *a, **kw: None)
|
|
# Presentation is unrelated to transactional business mutations under test.
|
|
monkeypatch.setattr(recharge, 'apply_to_read', lambda row, *a, **kw: row)
|
|
monkeypatch.setattr(recharge_approval, 'apply_to_read', lambda row, *a, **kw: row)
|
|
return (lambda uid: BxUserContext(uid, uid, 'd1', 'admin', 't1')), WechatNotice, WechatApprovalCycle
|
|
|
|
|
|
def bx_service(kind):
|
|
from app.services.bx.advance import AdvanceService
|
|
from app.services.bx.reimburse import ReimburseService
|
|
return AdvanceService if kind == 'bx_advance' else ReimburseService
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse'])
|
|
def test_bx_real_submit_sequential_approve_notifies_applicant_and_rolls_back(db, live, kind):
|
|
import json
|
|
user, Notice, Cycle = live
|
|
row = seed(db, kind, status='draft')
|
|
db.commit()
|
|
service = bx_service(kind)
|
|
service.submit(db, row.id, user('operator'))
|
|
assert db.query(Notice).filter_by(status='pending').one().recipient_id == 'u1'
|
|
service.approve(db, row.id, user('u1'), '同意')
|
|
assert db.query(Notice).filter_by(status='pending').one().recipient_id == 'u2'
|
|
service.approve(db, row.id, user('u2'), '同意')
|
|
result = db.query(Notice).filter_by(status='pending').one()
|
|
assert result.kind == 'result'
|
|
assert result.recipient_id == 'applicant'
|
|
assert json.loads(result.payload)['thing6']['value'] == 'u2'
|
|
assert json.loads(result.payload)['const5']['value'] == '通过'
|
|
db.rollback()
|
|
assert db.query(Notice).count() == 0
|
|
assert db.query(Cycle).count() == 0
|
|
assert row.status == 'draft'
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse'])
|
|
def test_bx_real_withdraw_resubmit_and_reject_use_new_cycle(db, live, kind):
|
|
user, Notice, Cycle = live
|
|
row = seed(db, kind, status='draft')
|
|
db.commit()
|
|
service = bx_service(kind)
|
|
service.submit(db, row.id, user('applicant'))
|
|
first_cycle = db.query(Cycle).one().cycle_id
|
|
service.withdraw(db, row.id, user('applicant'))
|
|
assert db.query(Notice).filter_by(status='pending').count() == 0
|
|
service.submit(db, row.id, user('operator'))
|
|
assert db.query(Cycle).one().cycle_id != first_cycle
|
|
service.reject(db, row.id, user('u1'), '请补充')
|
|
result = db.query(Notice).filter_by(status='pending').one()
|
|
assert result.kind == 'result'
|
|
assert result.result == 'rejected'
|
|
assert result.recipient_id == 'applicant'
|
|
service.submit(db, row.id, user('applicant'))
|
|
assert db.query(Notice).filter_by(status='pending').one().kind == 'pending'
|
|
|
|
|
|
def test_recharge_real_approve_handoff_withdraw_resubmit_reject_cancel(db, live):
|
|
from app.services import pl_vehicle_recharge_service as service
|
|
from app.services import pl_vehicle_recharge_approval_service as approval_service
|
|
user, Notice, Cycle = live
|
|
row = seed(db, 'pl_recharge', status='draft')
|
|
db.commit()
|
|
service.submit_apply(db, 't1', user('operator'), row.id)
|
|
assert db.query(Notice).filter_by(status='pending').one().recipient_id == 'u1'
|
|
first_cycle = db.query(Cycle).one().cycle_id
|
|
service.withdraw_apply(db, 't1', user('applicant'), row.id)
|
|
assert db.query(Notice).filter_by(status='pending').count() == 0
|
|
service.submit_apply(db, 't1', user('applicant'), row.id)
|
|
assert db.query(Cycle).one().cycle_id != first_cycle
|
|
approval_service.approve_apply(db, 't1', user('u1'), row.id)
|
|
result = db.query(Notice).filter_by(status='pending').one()
|
|
assert result.kind == 'result'
|
|
assert result.result == 'approved'
|
|
assert result.recipient_id == 'applicant'
|
|
assert row.status == 'in_approval'
|
|
service.withdraw_apply(db, 't1', user('applicant'), row.id)
|
|
service.submit_apply(db, 't1', user('applicant'), row.id)
|
|
approval_service.reject_apply(db, 't1', user('u1'), row.id, opinion='请补充')
|
|
assert db.query(Notice).filter_by(status='pending').one().result == 'rejected'
|
|
service.cancel_apply(db, 't1', user('operator'), row.id)
|
|
assert db.query(Notice).filter_by(status='pending').count() == 0
|
|
|
|
|
|
def test_recharge_real_direct_disburse_submission_has_no_pending_or_result(db, live):
|
|
from app.services import pl_vehicle_recharge_service as service
|
|
user, Notice, Cycle = live
|
|
row = seed(db, 'pl_recharge', status='draft', disburse_first=True)
|
|
db.commit()
|
|
service.submit_apply(db, 't1', user('operator'), row.id)
|
|
assert db.query(Cycle).count() == 1
|
|
assert db.query(Notice).count() == 0
|
|
|
|
|
|
def test_recharge_blank_action_is_approval_like_business_service(db):
|
|
seed(db, 'pl_recharge')
|
|
db.query(BxWorkflowNode).filter_by(id='n1').one().action_type = ' '
|
|
db.flush()
|
|
assert [p.user_id for p in snapshot(db, 'pl_recharge').pending] == ['u1']
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse', 'pl_recharge'])
|
|
def test_deleted_and_other_tenant_nodes_never_produce_tasks(db, kind):
|
|
seed(db, kind)
|
|
node = db.query(BxWorkflowNode).filter_by(id='n1').one()
|
|
node.tenant_no = 'other'
|
|
db.flush()
|
|
assert snapshot(db, kind).pending == []
|
|
node.tenant_no = 't1'
|
|
node.del_flag = 1
|
|
db.flush()
|
|
assert snapshot(db, kind).pending == []
|
|
|
|
|
|
@pytest.mark.parametrize('direct', [False, True])
|
|
def test_real_disbursement_preserves_approval_result_and_adds_completion(db, live, direct):
|
|
import json
|
|
from app.models.pl_models import PlVehicleRechargeRecord
|
|
from app.schemas.pl.vehicle_recharge import VehicleRechargeRecordCreate
|
|
from app.services import pl_vehicle_recharge_service as service
|
|
from app.services import pl_vehicle_recharge_approval_service as approval_service
|
|
from app.services.wechat_approval_notices import _valid_now
|
|
user, Notice, _ = live
|
|
PlVehicleRechargeRecord.__table__.create(db.get_bind())
|
|
row = seed(db, 'pl_recharge', status='draft', disburse_first=direct)
|
|
db.commit()
|
|
service.submit_apply(db, 't1', user('operator'), row.id)
|
|
if not direct:
|
|
approval_service.approve_apply(db, 't1', user('u1'), row.id)
|
|
approval_service.disburse_apply(
|
|
db, 't1', user('u1' if direct else 'u2'), row.id,
|
|
VehicleRechargeRecordCreate(rechargeDate=date.today(), rechargeType='fuel', amount=100),
|
|
)
|
|
assert row.status == 'disbursed'
|
|
completion = db.query(Notice).filter_by(status='pending', kind='completion').one()
|
|
assert json.loads(completion.payload)['thing3']['value'] == '充值发放'
|
|
assert completion.recipient_id == 'applicant'
|
|
assert _valid_now(db, completion) is True
|
|
pending = db.query(Notice).filter_by(status='pending').filter(
|
|
Notice.kind.in_(('pending', 'result'))
|
|
).all()
|
|
if direct:
|
|
assert pending == []
|
|
else:
|
|
assert len(pending) == 1
|
|
assert pending[0].kind == 'result'
|
|
assert json.loads(pending[0].payload)['thing6']['value'] == 'u1'
|
|
assert _valid_now(db, pending[0]) is True
|
|
|
|
|
|
def test_recharge_notification_failure_occurs_before_original_commit(db, live, monkeypatch):
|
|
from app.services import pl_vehicle_recharge_service as service
|
|
user, _, _ = live
|
|
row = seed(db, 'pl_recharge', status='draft')
|
|
db.commit()
|
|
original_round = row.approval_round
|
|
def failed_enqueue(*args, **kwargs):
|
|
raise RuntimeError('transactional enqueue failed')
|
|
monkeypatch.setattr(service, 'sync_approval_notifications', failed_enqueue)
|
|
with pytest.raises(RuntimeError, match='transactional enqueue failed'):
|
|
service.submit_apply(db, 't1', user('operator'), row.id)
|
|
db.rollback()
|
|
assert row.status == 'draft'
|
|
assert row.approval_round == original_round
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse'])
|
|
@pytest.mark.parametrize('operation', ['approve', 'reject'])
|
|
def test_bx_rejects_nonpending_state_even_with_leftover_node(db, live, kind, operation):
|
|
user, Notice, _ = live
|
|
row = seed(db, kind, status='draft')
|
|
db.commit()
|
|
assert getattr(bx_service(kind), operation)(db, row.id, user('u1'), '重复请求') is None
|
|
assert row.status == 'draft'
|
|
assert db.query(Notice).count() == 0
|
|
approval_model = BxAdvanceApproval if kind == 'bx_advance' else BxReimburseApproval
|
|
assert db.query(approval_model).count() == 0
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse'])
|
|
def test_bx_duplicate_submit_does_not_restart_active_cycle(db, live, kind):
|
|
user, Notice, Cycle = live
|
|
row = seed(db, kind, status='draft')
|
|
db.commit()
|
|
service = bx_service(kind)
|
|
service.submit(db, row.id, user('applicant'))
|
|
first_cycle = db.query(Cycle).one().cycle_id
|
|
assert service.submit(db, row.id, user('applicant')) is None
|
|
assert db.query(Cycle).one().cycle_id == first_cycle
|
|
assert db.query(Notice).count() == 1
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse'])
|
|
def test_bx_duplicate_terminal_approval_refreshes_cached_state(db, live, kind):
|
|
from sqlalchemy import update
|
|
user, Notice, _ = live
|
|
row = seed(db, kind)
|
|
row.current_node = '审批2'
|
|
db.commit()
|
|
assert row.current_node == '审批2'
|
|
# 另一事务已完成审批；当前Session仍持有旧的identity-map对象。
|
|
db.execute(update(type(row)).where(type(row).id == row.id).values(
|
|
status='approved', current_node=None
|
|
).execution_options(synchronize_session=False))
|
|
assert row.status == 'pending'
|
|
assert bx_service(kind).approve(db, row.id, user('u2'), '重复终审') is None
|
|
assert row.status == 'approved'
|
|
approval_model = BxAdvanceApproval if kind == 'bx_advance' else BxReimburseApproval
|
|
assert db.query(approval_model).count() == 0
|
|
assert db.query(Notice).count() == 0
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse', 'pl_recharge'])
|
|
def test_submit_locks_business_row_before_state_change(db, live, kind):
|
|
from sqlalchemy import event
|
|
from sqlalchemy.dialects import mysql
|
|
from app.services import pl_vehicle_recharge_service as recharge
|
|
user, _, _ = live
|
|
row = seed(db, kind, status='draft')
|
|
db.commit()
|
|
row_id = row.id
|
|
selects = []
|
|
locked_before_pending_write = []
|
|
def capture(orm_state):
|
|
if orm_state.is_select:
|
|
sql = str(orm_state.statement.compile(dialect=mysql.dialect()))
|
|
if f'FROM {row.__tablename__} ' in sql or f'FROM {row.__tablename__}\n' in sql:
|
|
selects.append(sql)
|
|
def before_flush(session, flush_context, instances):
|
|
if row in session.dirty and row.status in ('pending', 'in_approval'):
|
|
locked_before_pending_write.append(any('FOR UPDATE' in sql for sql in selects))
|
|
event.listen(db, 'do_orm_execute', capture)
|
|
event.listen(db, 'before_flush', before_flush)
|
|
if kind == 'pl_recharge':
|
|
recharge.submit_apply(db, 't1', user('applicant'), row_id)
|
|
else:
|
|
bx_service(kind).submit(db, row_id, user('applicant'))
|
|
# A read-only lookup may precede the definition lock. The authoritative
|
|
# business-row read must still lock and refresh before the pending write.
|
|
assert locked_before_pending_write and all(locked_before_pending_write)
|
|
|
|
|
|
def test_recharge_duplicate_terminal_approval_refreshes_cached_state(db, live):
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import update
|
|
from app.services import pl_vehicle_recharge_approval_service as service
|
|
user, Notice, _ = live
|
|
row = seed(db, 'pl_recharge')
|
|
db.commit()
|
|
assert row.current_node == 'n1'
|
|
db.execute(update(type(row)).where(type(row).id == row.id).values(
|
|
status='rejected', current_node=None
|
|
).execution_options(synchronize_session=False))
|
|
assert row.status == 'in_approval'
|
|
with pytest.raises(HTTPException) as exc:
|
|
service.approve_apply(db, 't1', user('u1'), row.id)
|
|
assert exc.value.status_code == 400
|
|
assert row.status == 'rejected'
|
|
assert db.query(PlVehicleRechargeApproval).count() == 0
|
|
assert db.query(Notice).count() == 0
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse', 'pl_recharge'])
|
|
def test_pending_uses_web_approval_instead_of_nonactionable_mobile_detail(db, kind):
|
|
seed(db, kind)
|
|
assert snapshot(db, kind).page == ''
|