shanxia2.0/backend/tests/test_wechat_notice_events.py

"""独立业务事件复用通知队列，保持外层事务及收件人隔离。"""
import importlib
import importlib.util
import json

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

from app.core.config import settings
from app.models.models import SysUser
from app.models.wechat_notice import WechatNotice


def service():
    name = 'app.services.wechat_notice_events'
    assert importlib.util.find_spec(name), '独立通知事件入队尚未实现'
    return importlib.import_module(name)


@pytest.fixture
def db(monkeypatch):
    engine = create_engine('sqlite://')
    # Python 3.9 sqlite驱动默认的旧事务模式不保证SAVEPOINT外层回滚。
    @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):
        model.__table__.create(engine)
    session = sessionmaker(bind=engine, autoflush=False)()
    session.add(SysUser(id='u1', tenant_no='t1', user_no='one', nick_name='申请人'))
    session.commit()
    monkeypatch.setattr(settings, 'WECHAT_OA_ENABLED', True)
    yield session
    session.close()
    engine.dispose()


def enqueue(db, **overrides):
    kwargs = dict(tenant_no='t1', business_type='bx_advance', business_id='doc1',
                  event_id='issuance1', kind='completion', recipient_id='u1',
                  template_id='template', payload={'thing3': {'value': '备用金发放'}})
    kwargs.update(overrides)
    return service().enqueue_event(db, **kwargs)


def test_disabled_never_touches_database(monkeypatch):
    monkeypatch.setattr(settings, 'WECHAT_OA_ENABLED', False)
    assert enqueue(None) is None


def test_unique_business_event_and_outer_rollback(db):
    row = enqueue(db)
    assert row.kind == 'completion'
    assert json.loads(row.payload)['thing3']['value'] == '备用金发放'
    assert enqueue(db) is None
    assert db.query(WechatNotice).count() == 1
    db.rollback()
    assert db.query(WechatNotice).count() == 0


def test_new_event_is_independent_of_approval_cycle(db):
    enqueue(db)
    enqueue(db, event_id='issuance2')
    db.commit()
    assert db.query(WechatNotice).count() == 2


def test_reject_other_tenant_disabled_and_deleted_recipient(db):
    assert enqueue(db, tenant_no='t2') is None
    user = db.query(SysUser).one()
    user.status = '0'
    db.flush()
    assert enqueue(db) is None
    user.status = '1'
    user.del_flag = 1
    db.flush()
    assert enqueue(db) is None


def test_invalid_payload_is_auditable_without_sending(db):
    row = enqueue(db, invalid_error='单号超过20字符')
    assert row.status == 'failed'
    assert row.last_error == '单号超过20字符'
    assert enqueue(db) is None


def test_missing_template_is_failed(db):
    row = enqueue(db, template_id='')
    assert row.status == 'failed'
    assert row.last_error


def test_unique_conflict_only_rolls_back_savepoint(db, monkeypatch):
    enqueue(db)
    db.commit()
    # 模拟并发者在预查询与INSERT之间已抢先写入相同事件。
    original_query = db.query
    class MissOnce:
        def __init__(self, query):
            self.query = query
        def filter_by(self, **kwargs):
            self.query = self.query.filter_by(**kwargs)
            return self
        def first(self):
            return None
    missed = []
    def query(*entities, **kwargs):
        result = original_query(*entities, **kwargs)
        if len(entities) == 1 and entities[0] is WechatNotice.id and not missed:
            missed.append(True)
            return MissOnce(result)
        return result
    monkeypatch.setattr(db, 'query', query)
    user = original_query(SysUser).one()
    user.nick_name = '业务事务的更新'
    assert enqueue(db) is None
    db.commit()
    assert original_query(WechatNotice).count() == 1
    assert original_query(SysUser).one().nick_name == '业务事务的更新'


def test_approval_resubmit_does_not_cancel_completion(db, monkeypatch):
    from app.models.wechat_notice import WechatApprovalCycle
    from app.services import wechat_approval_notices as approval
    from app.services.wechat_notice_types import NoticeSnapshot, PendingApprover
    WechatApprovalCycle.__table__.create(db.get_bind())
    row = enqueue(db)
    snapshot = NoticeSnapshot('备用金审批', 'BYJ001', 'u1', [PendingApprover('n1', 'u1')])
    monkeypatch.setattr(approval, '_snapshot', lambda *args: snapshot)
    approval.sync_approval_notifications(db, 'bx_advance', 'doc1', 't1', new_cycle=True)
    assert row.status == 'pending'
    approval.sync_approval_notifications(db, 'bx_advance', 'doc1', 't1')
    assert row.status == 'pending'


@pytest.mark.parametrize('kind,module,function', [
    ('completion', 'wechat_completion_notices', 'valid_completion_notice'),
    ('cert_daily', 'wechat_cert_notices', 'refresh_cert_notice'),
])
def test_worker_dispatches_independent_events_without_approval_cycle(db, monkeypatch, kind, module, function):
    import sys
    from types import SimpleNamespace
    from app.models.wechat_notice import WechatBinding
    from app.services import wechat_approval_notices as worker
    WechatBinding.__table__.create(db.get_bind())
    db.add(WechatBinding(id='bind', tenant_no='t1', user_id='u1', appid=settings.WECHAT_OA_APPID,
                         openid='service-openid', subscribed=1))
    row = enqueue(db, kind=kind)
    db.commit()
    checked, sent = [], []
    def validate(session, notice):
        checked.append(notice.id)
        return True
    monkeypatch.setitem(sys.modules, f'app.services.{module}', SimpleNamespace(**{function: validate}))
    client = SimpleNamespace(send=lambda **kwargs: sent.append(kwargs) or 'message-id')
    assert worker.deliver_pending(db, client=client) == 1
    assert checked == [row.id]
    assert sent[0]['openid'] == 'service-openid'
    assert worker.deliver_pending(db, client=client) == 0


def test_cert_daily_uses_existing_reminder_list(db, monkeypatch):
    from app.services.wechat_approval_notices import _fallback_url
    monkeypatch.setattr(settings, 'WECHAT_OA_PUBLIC_BASE_URL', 'https://portal.example')
    row = enqueue(db, kind='cert_daily', business_type='pl_cert_expiry')
    assert _fallback_url(row) == 'https://portal.example/#/pl/document-reminder-module/document-reminder'