"""真实业务事务中的完成通知；不连接生产、不调用微信。""" 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)