test_wechat_notice_events.py 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172
  1. """独立业务事件复用通知队列,保持外层事务及收件人隔离。"""
  2. import importlib
  3. import importlib.util
  4. import json
  5. import pytest
  6. from sqlalchemy import create_engine, event
  7. from sqlalchemy.orm import sessionmaker
  8. from app.core.config import settings
  9. from app.models.models import SysUser
  10. from app.models.wechat_notice import WechatNotice
  11. def service():
  12. name = 'app.services.wechat_notice_events'
  13. assert importlib.util.find_spec(name), '独立通知事件入队尚未实现'
  14. return importlib.import_module(name)
  15. @pytest.fixture
  16. def db(monkeypatch):
  17. engine = create_engine('sqlite://')
  18. # Python 3.9 sqlite驱动默认的旧事务模式不保证SAVEPOINT外层回滚。
  19. @event.listens_for(engine, 'connect')
  20. def connect(connection, _):
  21. connection.isolation_level = None
  22. @event.listens_for(engine, 'begin')
  23. def begin(connection):
  24. connection.exec_driver_sql('BEGIN')
  25. for model in (SysUser, WechatNotice):
  26. model.__table__.create(engine)
  27. session = sessionmaker(bind=engine, autoflush=False)()
  28. session.add(SysUser(id='u1', tenant_no='t1', user_no='one', nick_name='申请人'))
  29. session.commit()
  30. monkeypatch.setattr(settings, 'WECHAT_OA_ENABLED', True)
  31. yield session
  32. session.close()
  33. engine.dispose()
  34. def enqueue(db, **overrides):
  35. kwargs = dict(tenant_no='t1', business_type='bx_advance', business_id='doc1',
  36. event_id='issuance1', kind='completion', recipient_id='u1',
  37. template_id='template', payload={'thing3': {'value': '备用金发放'}})
  38. kwargs.update(overrides)
  39. return service().enqueue_event(db, **kwargs)
  40. def test_disabled_never_touches_database(monkeypatch):
  41. monkeypatch.setattr(settings, 'WECHAT_OA_ENABLED', False)
  42. assert enqueue(None) is None
  43. def test_unique_business_event_and_outer_rollback(db):
  44. row = enqueue(db)
  45. assert row.kind == 'completion'
  46. assert json.loads(row.payload)['thing3']['value'] == '备用金发放'
  47. assert enqueue(db) is None
  48. assert db.query(WechatNotice).count() == 1
  49. db.rollback()
  50. assert db.query(WechatNotice).count() == 0
  51. def test_new_event_is_independent_of_approval_cycle(db):
  52. enqueue(db)
  53. enqueue(db, event_id='issuance2')
  54. db.commit()
  55. assert db.query(WechatNotice).count() == 2
  56. def test_reject_other_tenant_disabled_and_deleted_recipient(db):
  57. assert enqueue(db, tenant_no='t2') is None
  58. user = db.query(SysUser).one()
  59. user.status = '0'
  60. db.flush()
  61. assert enqueue(db) is None
  62. user.status = '1'
  63. user.del_flag = 1
  64. db.flush()
  65. assert enqueue(db) is None
  66. def test_invalid_payload_is_auditable_without_sending(db):
  67. row = enqueue(db, invalid_error='单号超过20字符')
  68. assert row.status == 'failed'
  69. assert row.last_error == '单号超过20字符'
  70. assert enqueue(db) is None
  71. def test_missing_template_is_failed(db):
  72. row = enqueue(db, template_id='')
  73. assert row.status == 'failed'
  74. assert row.last_error
  75. def test_unique_conflict_only_rolls_back_savepoint(db, monkeypatch):
  76. enqueue(db)
  77. db.commit()
  78. # 模拟并发者在预查询与INSERT之间已抢先写入相同事件。
  79. original_query = db.query
  80. class MissOnce:
  81. def __init__(self, query):
  82. self.query = query
  83. def filter_by(self, **kwargs):
  84. self.query = self.query.filter_by(**kwargs)
  85. return self
  86. def first(self):
  87. return None
  88. missed = []
  89. def query(*entities, **kwargs):
  90. result = original_query(*entities, **kwargs)
  91. if len(entities) == 1 and entities[0] is WechatNotice.id and not missed:
  92. missed.append(True)
  93. return MissOnce(result)
  94. return result
  95. monkeypatch.setattr(db, 'query', query)
  96. user = original_query(SysUser).one()
  97. user.nick_name = '业务事务的更新'
  98. assert enqueue(db) is None
  99. db.commit()
  100. assert original_query(WechatNotice).count() == 1
  101. assert original_query(SysUser).one().nick_name == '业务事务的更新'
  102. def test_approval_resubmit_does_not_cancel_completion(db, monkeypatch):
  103. from app.models.wechat_notice import WechatApprovalCycle
  104. from app.services import wechat_approval_notices as approval
  105. from app.services.wechat_notice_types import NoticeSnapshot, PendingApprover
  106. WechatApprovalCycle.__table__.create(db.get_bind())
  107. row = enqueue(db)
  108. snapshot = NoticeSnapshot('备用金审批', 'BYJ001', 'u1', [PendingApprover('n1', 'u1')])
  109. monkeypatch.setattr(approval, '_snapshot', lambda *args: snapshot)
  110. approval.sync_approval_notifications(db, 'bx_advance', 'doc1', 't1', new_cycle=True)
  111. assert row.status == 'pending'
  112. approval.sync_approval_notifications(db, 'bx_advance', 'doc1', 't1')
  113. assert row.status == 'pending'
  114. @pytest.mark.parametrize('kind,module,function', [
  115. ('completion', 'wechat_completion_notices', 'valid_completion_notice'),
  116. ('cert_daily', 'wechat_cert_notices', 'refresh_cert_notice'),
  117. ])
  118. def test_worker_dispatches_independent_events_without_approval_cycle(db, monkeypatch, kind, module, function):
  119. import sys
  120. from types import SimpleNamespace
  121. from app.models.wechat_notice import WechatBinding
  122. from app.services import wechat_approval_notices as worker
  123. WechatBinding.__table__.create(db.get_bind())
  124. db.add(WechatBinding(id='bind', tenant_no='t1', user_id='u1', appid=settings.WECHAT_OA_APPID,
  125. openid='service-openid', subscribed=1))
  126. row = enqueue(db, kind=kind)
  127. db.commit()
  128. checked, sent = [], []
  129. def validate(session, notice):
  130. checked.append(notice.id)
  131. return True
  132. monkeypatch.setitem(sys.modules, f'app.services.{module}', SimpleNamespace(**{function: validate}))
  133. client = SimpleNamespace(send=lambda **kwargs: sent.append(kwargs) or 'message-id')
  134. assert worker.deliver_pending(db, client=client) == 1
  135. assert checked == [row.id]
  136. assert sent[0]['openid'] == 'service-openid'
  137. assert worker.deliver_pending(db, client=client) == 0
  138. def test_cert_daily_uses_existing_reminder_list(db, monkeypatch):
  139. from app.services.wechat_approval_notices import _fallback_url
  140. monkeypatch.setattr(settings, 'WECHAT_OA_PUBLIC_BASE_URL', 'https://portal.example')
  141. row = enqueue(db, kind='cert_daily', business_type='pl_cert_expiry')
  142. assert _fallback_url(row) == 'https://portal.example/#/pl/document-reminder-module/document-reminder'