shanxia2.0/backend/tests/test_wechat_completion_notices.py

"""真实业务事务中的完成通知；不连接生产、不调用微信。"""
from datetime import date, datetime
import importlib
import importlib.util
import json

import pytest
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker

from app.core.bx_auth import BxUserContext
from app.core.config import settings
from app.models.bx_models import BxAdvanceApply, BxReimburseApply, BxReimburseOffset, BxWorkflowNode
from app.models.models import SysUser
from app.models.pl_models import (
    PlFreightSettlement, PlFreightSettlementApproval, PlRemittanceVoucher,
    PlVehicleRechargeApply, PlVehicleRechargeApproval, PlVehicleRechargeRecord,
)
from app.models.wechat_notice import WechatNotice

NOW = datetime(2026, 10, 27, 12, 30, 45)


def service():
    name = 'app.services.wechat_completion_notices'
    assert importlib.util.find_spec(name), '完成通知尚未实现'
    return importlib.import_module(name)


@pytest.fixture
def db(monkeypatch):
    engine = create_engine('sqlite://')
    @event.listens_for(engine, 'connect')
    def connect(connection, _):
        connection.isolation_level = None
    @event.listens_for(engine, 'begin')
    def begin(connection):
        connection.exec_driver_sql('BEGIN')
    for model in (SysUser, WechatNotice, BxAdvanceApply, BxReimburseApply,
                  BxReimburseOffset, BxWorkflowNode, PlFreightSettlement,
                  PlFreightSettlementApproval, PlRemittanceVoucher,
                  PlVehicleRechargeApply, PlVehicleRechargeApproval, PlVehicleRechargeRecord):
        model.__table__.create(engine)
    session = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)()
    for uid in ('applicant', 'operator', 'submitter'):
        session.add(SysUser(id=uid, user_no=uid, nick_name=uid, tenant_no='t1'))
    session.commit()
    monkeypatch.setattr(settings, 'WECHAT_OA_ENABLED', True)
    monkeypatch.setattr(settings, 'WECHAT_OA_TEMPLATE_COMPLETION', 'completion-template')
    yield session
    session.close()
    engine.dispose()


def seed(db, kind, *, completed=True, biz_id='doc'):
    common = dict(id=biz_id, tenant_no='t1', create_by='operator')
    if kind in ('bx_advance', 'bx_reimburse'):
        common.update(apply_code='BYJ20261027001' if kind == 'bx_advance' else 'BX20261027001',
                      applicant_id='applicant', applicant_name='申请人', amount=100)
        if kind == 'bx_advance':
            row = BxAdvanceApply(**common, reason='备用金', status='issued' if completed else 'approved',
                                 issue_status='issued' if completed else None, issue_time=NOW if completed else None)
        else:
            row = BxReimburseApply(**common, title='报销', reimburse_date=date.today(), expense_id='expense',
                                   expense_name='费用', status='processed' if completed else 'approved',
                                   pay_status='paid' if completed else None, pay_time=NOW if completed else None)
    elif kind == 'pl_freight':
        row = PlFreightSettlement(**common, settlement_no='YF2026102701', total_freight=100,
                                  settlement_status='confirmed' if completed else 'approved')
        db.add(PlFreightSettlementApproval(id='approval', tenant_no='t1', settlement_id=biz_id,
               step=1, approver_id='operator', create_by='submitter', approval_result='approved'))
        if completed:
            db.add(PlRemittanceVoucher(id='record', tenant_no='t1', settlement_id=biz_id,
                   remittance_date=date(2020, 1, 1), create_time=NOW))
    else:
        row = PlVehicleRechargeApply(**common, apply_no='ETC2026102701' + biz_id, applicant_id='applicant',
                                     license_plate='鄂A00001', recharge_type='etc', apply_date=date.today(),
                                     status='disbursed' if completed else 'pending')
        if completed:
            db.add(PlVehicleRechargeRecord(id='record', tenant_no='t1', apply_id=biz_id,
                   recharge_order_no='SXCL20261027000001', recharge_date=date(2020, 1, 1),
                   recharge_type='etc', license_plate='鄂A00001', amount=100, create_time=NOW))
    db.add(row)
    db.commit()
    return row


def sync(db, kind, **kwargs):
    return service().sync_completion_notification(db, kind, 'doc', 't1', actor_id='operator', **kwargs)


@pytest.mark.parametrize('kind,name,state', [
    ('bx_advance', '备用金发放', '发放完成'), ('bx_reimburse', '报销支付', '支付完成'),
    ('pl_freight', '运费发放', '发放完成'), ('pl_recharge', '充值发放', '发放完成'),
])
def test_exact_payload_real_time_recipient_without_approval_cycle_and_rollback(db, kind, name, state):
    seed(db, kind)
    kwargs = {'event_id': 'record', 'occurred_at': NOW} if kind.startswith('pl_') else {}
    sync(db, kind, **kwargs)
    notice = db.query(WechatNotice).one()
    assert notice.recipient_id == ('submitter' if kind == 'pl_freight' else 'applicant')
    assert set(json.loads(notice.payload)) == {'thing3', 'thing4', 'thing5', 'time12'}
    assert json.loads(notice.payload)['thing3']['value'] == name
    assert json.loads(notice.payload)['thing5']['value'] == state
    assert json.loads(notice.payload)['time12']['value'] == '2026-10-27 12:30:45'
    assert service().valid_completion_notice(db, notice)
    sync(db, kind, **kwargs)
    assert db.query(WechatNotice).count() == 1
    db.rollback()
    assert db.query(WechatNotice).count() == 0


@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse', 'pl_freight', 'pl_recharge'])
def test_status_only_approval_never_generates_completion(db, kind):
    seed(db, kind, completed=False)
    sync(db, kind, event_id='record', occurred_at=NOW)
    assert db.query(WechatNotice).count() == 0


def test_full_number_is_not_truncated_and_invalid_records_are_auditable(db):
    row = seed(db, 'bx_advance')
    row.apply_code = 'X' * 21
    db.flush()
    sync(db, 'bx_advance')
    notice = db.query(WechatNotice).one()
    assert notice.status == 'failed'
    assert json.loads(notice.payload)['thing4']['value'] == 'X' * 21
    assert notice.last_error


def test_recipient_and_tenant_or_deleted_document_revalidated(db):
    row = seed(db, 'bx_advance')
    sync(db, 'bx_advance')
    notice = db.query(WechatNotice).one()
    row.tenant_no = 'other'
    db.flush()
    assert not service().valid_completion_notice(db, notice)
    row.tenant_no = 't1'
    row.del_flag = 1
    db.flush()
    assert not service().valid_completion_notice(db, notice)
    row.del_flag = 0
    row.applicant_id = 'operator'
    db.flush()
    assert not service().valid_completion_notice(db, notice)


@pytest.fixture
def business(db, monkeypatch):
    from app.services.bx import base
    from app.services.bx.advance import AdvanceService
    from app.services.bx.reimburse import ReimburseService
    from app.services import pl_vehicle_recharge_service as recharge
    from app.services import pl_vehicle_recharge_approval_service as approval
    monkeypatch.setattr(base, 'assert_bx_row_in_scope', lambda *a, **kw: None)
    for cls in (AdvanceService, ReimburseService):
        monkeypatch.setattr(cls, '_tenant_no', lambda *a: 't1')
    monkeypatch.setattr(recharge, 'assert_pl_row_in_scope', lambda *a, **kw: None)
    monkeypatch.setattr(recharge, 'record_to_read', lambda row, *a, **kw: row)
    monkeypatch.setattr(recharge, 'apply_to_read', lambda row, *a, **kw: row)
    monkeypatch.setattr(approval, 'apply_to_read', lambda row, *a, **kw: row)
    return BxUserContext('operator', '经办人', 'dept', 'admin', 't1')


@pytest.mark.parametrize('kind', ['bx_advance', 'bx_reimburse'])
def test_real_bx_complete_hook_retry_preserves_original_time_and_rolls_back(db, business, kind):
    from app.services.bx.advance import AdvanceService
    from app.services.bx.reimburse import ReimburseService
    row = seed(db, kind, completed=False)
    fn = AdvanceService.issue if kind == 'bx_advance' else ReimburseService.pay
    fn(db, row.id, business)
    assert db.query(WechatNotice).count() == 1
    payload = db.query(WechatNotice).one().payload
    stamp = row.issue_time if kind == 'bx_advance' else row.pay_time
    fn(db, row.id, business)
    assert (row.issue_time if kind == 'bx_advance' else row.pay_time) == stamp
    assert db.query(WechatNotice).count() == 1
    assert db.query(WechatNotice).one().payload == payload
    db.rollback()
    assert db.query(WechatNotice).count() == 0
    assert (row.issue_time if kind == 'bx_advance' else row.pay_time) is None


def test_batch_offset_only_notifies_real_payment_and_deduplicates_offsets(db, business):
    from app.services.bx.reimburse import ReimburseService
    row = seed(db, 'bx_reimburse', completed=False)
    items = [{'advance_id': 'advance', 'offset_amount': 30}]
    ReimburseService.batch_offset(db, row.id, items, 70, business)
    assert db.query(WechatNotice).count() == 1
    ReimburseService.batch_offset(db, row.id, items, 70, business)
    assert db.query(BxReimburseOffset).count() == 1
    db.rollback()
    ReimburseService.batch_offset(db, row.id, items, 0, business)
    assert db.query(WechatNotice).count() == 0
    assert row.pay_time is None


def test_freight_real_disbursement_notifies_submitter_with_voucher_time(db, business):
    from app.services.pl_freight_settlement_service import disburse_settlement
    from app.schemas.pl.freight import FreightSettlementDisburseBody
    row = seed(db, 'pl_freight', completed=False)
    disburse_settlement(db, tenant_no='t1', settlement_id=row.id,
                       body=FreightSettlementDisburseBody(remittance_date=date(2020, 1, 1), remittance_amount=100, voucher_urls=['proof']),
                       user_id='operator', dept_id='dept')
    notice = db.query(WechatNotice).one()
    voucher = db.query(PlRemittanceVoucher).one()
    assert row.settlement_status == 'confirmed'
    assert notice.recipient_id == 'submitter'
    assert notice.cycle_id == voucher.id
    assert json.loads(notice.payload)['time12']['value'] == voucher.create_time.strftime('%Y-%m-%d %H:%M:%S')
    assert service().valid_completion_notice(db, notice)
    voucher.del_flag = 1
    db.flush()
    assert not service().valid_completion_notice(db, notice)


def payload(apply_id='doc'):
    from app.schemas.pl.vehicle_recharge import VehicleRechargeRecordCreate
    return VehicleRechargeRecordCreate(apply_id=apply_id, license_plate='鄂A00001', recharge_type='etc',
                                       recharge_date=date(2020, 1, 1), amount=100)


def test_recharge_create_delete_reissue_and_no_applicant(db, business, monkeypatch):
    from app.services import pl_vehicle_recharge_service as recharge
    row = seed(db, 'pl_recharge', completed=False)
    record = recharge.create_record(db, 't1', business, payload())
    notice = db.query(WechatNotice).one()
    assert service().valid_completion_notice(db, notice)
    recharge.delete_records(db, 't1', business, [record.id])
    assert notice.status == 'cancelled'
    assert not service().valid_completion_notice(db, notice)
    recharge.create_record(db, 't1', business, payload())
    assert db.query(WechatNotice).filter_by(status='pending').count() == 1
    assert db.query(WechatNotice).count() == 2
    monkeypatch.setattr(recharge, '_assert_vehicle_exists', lambda *a: None)
    recharge.create_record(db, 't1', business, payload(None))
    assert db.query(WechatNotice).count() == 2


@pytest.mark.parametrize('old_status', ['pending', 'sending'])
def test_recharge_same_second_reassociation_has_independent_event_and_cancels_old(db, business, monkeypatch, old_status):
    from app.services import pl_vehicle_recharge_service as recharge
    class FrozenDateTime(datetime):
        @classmethod
        def now(cls, tz=None):
            return NOW
    monkeypatch.setattr(recharge, 'datetime', FrozenDateTime)
    first = seed(db, 'pl_recharge', completed=False)
    second = seed(db, 'pl_recharge', completed=False, biz_id='other')
    record = recharge.create_record(db, 't1', business, payload())
    original = db.query(WechatNotice).one()
    original.status = old_status
    db.commit()
    recharge.update_record(db, 't1', business, record.id, payload(second.id))
    recharge.update_record(db, 't1', business, record.id, payload(first.id))
    assert db.query(WechatNotice).count() == 3
    assert db.query(WechatNotice).filter_by(status='pending').count() == 1
    assert not service().valid_completion_notice(db, original)
    assert service().valid_completion_notice(db, db.query(WechatNotice).filter_by(status='pending').one())


def test_recharge_approval_disbursement_hook(db, business):
    from app.services.pl_vehicle_recharge_approval_service import disburse_apply
    row = seed(db, 'pl_recharge', completed=False)
    row.status, row.current_node, row.wf_id = 'in_approval', 'node', 'wf'
    db.add(BxWorkflowNode(id='node', wf_id='wf', node_name='发放', node_order=1,
                         approver_id='operator', action_type='disburse', tenant_no='t1'))
    db.commit()
    disburse_apply(db, 't1', business, row.id, payload())
    notice = db.query(WechatNotice).one()
    assert service().valid_completion_notice(db, notice)
    assert notice.recipient_id == 'applicant'


def test_recharge_disabled_still_cancels_old_pending_but_does_not_enqueue(db, business, monkeypatch):
    from app.services import pl_vehicle_recharge_service as recharge
    seed(db, 'pl_recharge', completed=False)
    record = recharge.create_record(db, 't1', business, payload())
    notice = db.query(WechatNotice).one()
    monkeypatch.setattr(settings, 'WECHAT_OA_ENABLED', False)
    recharge.delete_records(db, 't1', business, [record.id])
    recharge.create_record(db, 't1', business, payload())
    assert notice.status == 'cancelled'
    assert db.query(WechatNotice).count() == 1


def test_disabled_before_notification_schema_install_does_not_block_business(db, business, monkeypatch):
    from app.services.bx.advance import AdvanceService
    row = seed(db, 'bx_advance', completed=False)
    WechatNotice.__table__.drop(db.get_bind())
    monkeypatch.setattr(settings, 'WECHAT_OA_ENABLED', False)
    AdvanceService.issue(db, row.id, business)
    assert row.status == 'issued'


def test_completed_before_launch_is_not_retroactively_notified(db, business):
    from app.services.bx.advance import AdvanceService
    row = seed(db, 'bx_advance')
    AdvanceService.issue(db, row.id, business)
    assert db.query(WechatNotice).count() == 0


def test_invalid_recipient_never_enqueues(db):
    row = seed(db, 'bx_advance')
    row.applicant_id = 'outsider'
    db.add(SysUser(id='outsider', user_no='outsider', nick_name='外租户', tenant_no='other'))
    db.flush()
    sync(db, 'bx_advance')
    assert db.query(WechatNotice).count() == 0
    row.applicant_id = 'applicant'
    db.query(SysUser).filter_by(id='applicant').one().status = '0'
    db.flush()
    sync(db, 'bx_advance')
    assert db.query(WechatNotice).count() == 0


def test_stale_sender_cannot_reactivate_revoked_recharge_association(db, business):
    from app.services import pl_vehicle_recharge_service as recharge
    first = seed(db, 'pl_recharge', completed=False)
    second = seed(db, 'pl_recharge', completed=False, biz_id='other')
    record = recharge.create_record(db, 't1', business, payload())
    original = db.query(WechatNotice).one()
    original.status = 'sending'
    db.commit()
    recharge.update_record(db, 't1', business, record.id, payload(second.id))
    recharge.update_record(db, 't1', business, record.id, payload(first.id))
    # 网络在撤销前已开始时，旧worker可能把瞬时失败重新写为pending。
    original.status = 'pending'
    db.flush()
    assert not service().valid_completion_notice(db, original)


def test_freight_disbursement_consumes_one_locked_current_read(db, business, monkeypatch):
    from app.services import pl_freight_settlement_service as freight
    from app.schemas.pl.freight import FreightSettlementDisburseBody
    row = seed(db, 'pl_freight', completed=False)
    reads = []
    @event.listens_for(db, 'do_orm_execute')
    def track_reads(execute_state):
        statement = execute_state.statement
        if execute_state.is_select and any(
            entry.get('entity') is PlFreightSettlement
            for entry in getattr(statement, 'column_descriptions', [])
        ):
            reads.append(statement._for_update_arg is not None)
    # 只隔离通知快照及展示刷新读取；真实发放状态、凭证与提交仍执行。
    monkeypatch.setattr(freight, 'sync_completion_notification', lambda *a, **kw: None)
    monkeypatch.setattr(db, 'refresh', lambda *a, **kw: None)
    freight.disburse_settlement(db, tenant_no='t1', settlement_id=row.id,
        body=FreightSettlementDisburseBody(remittance_date=date.today(), remittance_amount=100, voucher_urls=['proof']),
        user_id='operator', dept_id='dept')
    assert reads == [True], '状态校验必须直接使用唯一锁读，不能随后退回普通快照读'


def test_freight_locked_get_refreshes_stale_identity_map(db):
    from app.services.pl_freight_settlement_service import get_settlement_row
    row = seed(db, 'pl_freight', completed=False)
    db.execute(PlFreightSettlement.__table__.update().where(
        PlFreightSettlement.id == row.id,
    ).values(settlement_status='confirmed'))
    assert row.settlement_status == 'approved'
    current = get_settlement_row(db, 't1', row.id, for_update=True)
    assert current is row
    assert current.settlement_status == 'confirmed'


def test_cancelled_freight_completion_is_not_revived(db):
    seed(db, 'pl_freight')
    notice = sync(db, 'pl_freight', event_id='record')
    assert notice is not None
    notice.status = 'cancelled'
    notice.result = 'revoked'
    db.flush()
    sync(db, 'pl_freight', event_id='record')
    assert db.query(WechatNotice).count() == 1
    assert notice.status == 'cancelled'
    assert not service().valid_completion_notice(db, notice)