shanxia2.0/backend/tests/test_wechat_approval_notices.py

from datetime import datetime, timedelta

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


@pytest.fixture
def notices(monkeypatch):
    from app.services import wechat_approval_notices as svc
    from app.services.wechat_notice_types import NoticeSnapshot, PendingApprover
    from app.models.wechat_notice import WechatApprovalCycle, WechatNotice, WechatBinding
    from app.models.models import SysUser
    engine = create_engine('sqlite://')
    for model in (SysUser, WechatApprovalCycle, WechatNotice, WechatBinding):
        model.__table__.create(engine)
    db = sessionmaker(bind=engine, autoflush=False)()
    for uid in ('applicant', 'a', 'b'):
        db.add(SysUser(id=uid, tenant_no='t1', user_no=uid, nick_name=uid))
    db.commit()
    monkeypatch.setattr(svc.settings, 'WECHAT_OA_ENABLED', True)
    snapshot = NoticeSnapshot('报销审批', 'BX001', 'applicant', [PendingApprover('n1', 'a')])
    monkeypatch.setattr(svc, '_snapshot', lambda *args: snapshot)
    yield svc, db, snapshot, WechatNotice, WechatBinding
    db.close()


def test_disabled_does_not_access_new_tables(monkeypatch):
    from app.services import wechat_approval_notices as svc
    monkeypatch.setattr(svc.settings, 'WECHAT_OA_ENABLED', False)
    svc.sync_approval_notifications(None, 'bx_reimburse', 'x', 't1', new_cycle=True)


def test_template_mapping_and_transactional_rollback(notices):
    svc, db, snapshot, Notice, _ = notices
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1', new_cycle=True)
    db.flush()
    row = db.query(Notice).one()
    import json
    assert json.loads(row.payload)['thing2']['value'] == '报销审批'
    assert json.loads(row.payload)['character_string6']['value'] == 'BX001'
    assert row.recipient_id == 'a'
    db.rollback()
    assert db.query(Notice).count() == 0


def test_dedup_next_step_result_and_resubmit(notices):
    svc, db, snapshot, Notice, _ = notices
    from app.services.wechat_notice_types import PendingApprover
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1', new_cycle=True)
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1')
    db.flush()
    assert db.query(Notice).count() == 1
    snapshot.pending = [PendingApprover('n2', 'b')]
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1')
    db.flush()
    assert db.query(Notice).filter_by(status='pending').one().recipient_id == 'b'
    snapshot.pending = []
    snapshot.result = 'rejected'
    snapshot.approver_id = 'b'
    snapshot.occurred_at = datetime(2026, 10, 27, 14, 0)
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1')
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1')
    db.flush()
    import json
    result = db.query(Notice).filter_by(kind='result').one()
    assert result.recipient_id == 'applicant'
    assert json.loads(result.payload)['const5']['value'] == '打回'
    assert json.loads(result.payload)['thing6']['value'] == 'b'
    snapshot.result = None
    snapshot.pending = [PendingApprover('n1', 'a')]
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1', new_cycle=True)
    db.flush()
    assert db.query(Notice).filter_by(status='pending').count() == 1


def test_no_history_backfill_and_tenant_recipient_guard(notices):
    svc, db, snapshot, Notice, _ = notices
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1')
    db.flush()
    assert db.query(Notice).count() == 0
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 'other', new_cycle=True)
    db.flush()
    assert db.query(Notice).count() == 0


class FakeClient:
    def __init__(self, failure=None):
        self.calls = []
        self.failure = failure

    def send(self, **kwargs):
        self.calls.append(kwargs)
        if self.failure:
            raise self.failure
        return 'wx-msg-id'


def bind(db, Binding):
    from app.core.config import settings
    db.add(Binding(id='binding', tenant_no='t1', user_id='a', appid=settings.WECHAT_OA_APPID,
                   openid='oa-openid', subscribed=1))
    db.commit()


def test_worker_rechecks_withdrawal_before_send(notices):
    svc, db, snapshot, Notice, Binding = notices
    bind(db, Binding)
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1', new_cycle=True)
    db.commit()
    snapshot.pending = []
    client = FakeClient()
    svc.deliver_pending(db, client=client)
    assert client.calls == []
    assert db.query(Notice).one().status == 'cancelled'


def test_worker_sends_once_with_oa_openid(notices):
    svc, db, snapshot, Notice, Binding = notices
    bind(db, Binding)
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1', new_cycle=True)
    db.commit()
    client = FakeClient()
    svc.deliver_pending(db, client=client)
    svc.deliver_pending(db, client=client)
    assert len(client.calls) == 1
    assert client.calls[0]['openid'] == 'oa-openid'
    assert db.query(Notice).one().status == 'sent'


def test_ambiguous_send_is_not_automatically_repeated(notices):
    svc, db, snapshot, Notice, Binding = notices
    from app.services.wechat_oa_client import WechatDeliveryUnknown
    bind(db, Binding)
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1', new_cycle=True)
    db.commit()
    client = FakeClient(WechatDeliveryUnknown('发送结果待核对'))
    svc.deliver_pending(db, client=client)
    svc.deliver_pending(db, client=client)
    assert len(client.calls) == 1
    assert db.query(Notice).one().status == 'unknown'


def test_explicit_transient_failure_is_bounded_retry(notices):
    svc, db, snapshot, Notice, Binding = notices
    from app.services.wechat_oa_client import WechatAPIError
    bind(db, Binding)
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1', new_cycle=True)
    db.commit()
    client = FakeClient(WechatAPIError(-1))
    for _ in range(6):
        svc.deliver_pending(db, client=client)
        row = db.query(Notice).one()
        row.next_attempt_at = datetime.now() - timedelta(seconds=1)
        db.commit()
    assert len(client.calls) == 5
    assert db.query(Notice).one().status == 'failed'


@pytest.mark.parametrize('biz_type,path', [
    ('bx_advance', '/bx/advance/approve'),
    ('bx_reimburse', '/bx/reimburse/approve'),
    ('pl_recharge', '/vehicle-mgmt/etc-recharge/vehicle-recharge-approval'),
])
def test_pending_links_to_real_approval_screen(monkeypatch, biz_type, path):
    from types import SimpleNamespace
    from app.services import wechat_approval_notices as svc
    monkeypatch.setattr(svc.settings, 'WECHAT_OA_PUBLIC_BASE_URL', 'https://portal.example.test')
    assert svc._fallback_url(SimpleNamespace(business_type=biz_type, kind='pending')) == 'https://portal.example.test/#' + path


def test_claimed_assignment_invalid_after_transfer_back(notices):
    svc, db, snapshot, Notice, _ = notices
    from app.services.wechat_notice_types import PendingApprover
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1', new_cycle=True)
    db.flush()
    old = db.query(Notice).one()
    old.status = 'sending'
    db.commit()
    snapshot.pending = [PendingApprover('n1', 'b')]
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1')
    snapshot.pending = [PendingApprover('n1', 'a')]
    svc.sync_approval_notifications(db, 'bx_reimburse', 'x', 't1')
    db.commit()
    assert not svc._valid_now(db, old)
    assert svc._valid_now(db, db.query(Notice).filter_by(status='pending', recipient_id='a').one())