Files
shanxia2.0/backend/tests/test_wechat_notice_events.py

172 lines
6.1 KiB
Python

"""独立业务事件复用通知队列,保持外层事务及收件人隔离。"""
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 not sent[0].get('url')
assert worker.deliver_pending(db, client=client) == 0
def test_cert_daily_has_no_mini_page(db):
row = enqueue(db, kind='cert_daily', business_type='pl_cert_expiry')
assert not row.page