shanxia2.0/backend/tests/test_wechat_notice_bx_sources.py

"""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 == ''