86 lines
3.9 KiB
Python
86 lines
3.9 KiB
Python
"""证件预警生成与微信摘要顺序、配置时区和失败隔离。"""
|
|
import sys
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
@pytest.mark.parametrize('enabled', [True, False])
|
|
def test_scan_commits_before_enqueue_and_disabled_skips(monkeypatch, enabled):
|
|
from app.tasks import pl_cert_expiry as task
|
|
events = []
|
|
db = SimpleNamespace(commit=lambda: events.append('commit'),
|
|
rollback=lambda: events.append('rollback'),
|
|
close=lambda: events.append('close'))
|
|
monkeypatch.setattr(task, 'SessionLocal', lambda: db)
|
|
monkeypatch.setattr(task, 'scan_cert_expiry', lambda session, **kwargs: events.append('scan') or 2)
|
|
monkeypatch.setattr(settings, 'WECHAT_OA_ENABLED', enabled)
|
|
monkeypatch.setattr(settings, 'WECHAT_OA_CERT_REMINDER_ENABLED', True)
|
|
module = SimpleNamespace(enqueue_daily_cert_notices=lambda session: events.append('enqueue') or 1)
|
|
monkeypatch.setitem(sys.modules, 'app.services.wechat_cert_notices', module)
|
|
task.scan_pl_cert_expiry_job()
|
|
assert events == (['scan', 'commit', 'enqueue', 'commit', 'close'] if enabled else
|
|
['scan', 'commit', 'close'])
|
|
|
|
|
|
def test_enqueue_failure_preserves_scanned_reminders(monkeypatch):
|
|
from app.tasks import pl_cert_expiry as task
|
|
events = []
|
|
db = SimpleNamespace(commit=lambda: events.append('commit'),
|
|
rollback=lambda: events.append('rollback'),
|
|
close=lambda: events.append('close'))
|
|
monkeypatch.setattr(task, 'SessionLocal', lambda: db)
|
|
monkeypatch.setattr(task, 'scan_cert_expiry', lambda session, **kwargs: events.append('scan') or 2)
|
|
monkeypatch.setattr(settings, 'WECHAT_OA_ENABLED', True)
|
|
monkeypatch.setattr(settings, 'WECHAT_OA_CERT_REMINDER_ENABLED', True)
|
|
def fail(session):
|
|
events.append('enqueue')
|
|
raise RuntimeError('queue unavailable')
|
|
monkeypatch.setitem(sys.modules, 'app.services.wechat_cert_notices',
|
|
SimpleNamespace(enqueue_daily_cert_notices=fail))
|
|
with pytest.raises(RuntimeError):
|
|
task.scan_pl_cert_expiry_job()
|
|
assert events == ['scan', 'commit', 'enqueue', 'rollback', 'close']
|
|
|
|
|
|
def test_daily_job_uses_explicit_shanghai_configured_time(monkeypatch):
|
|
import app.tasks as tasks
|
|
jobs = {}
|
|
scheduler = SimpleNamespace(running=False, start=lambda: None,
|
|
add_job=lambda function, trigger, **kwargs: jobs.update({kwargs['id']: kwargs}))
|
|
monkeypatch.setattr(tasks, 'scheduler', scheduler)
|
|
monkeypatch.setattr(settings, 'WECHAT_OA_CERT_REMINDER_HOUR', 9)
|
|
monkeypatch.setattr(settings, 'WECHAT_OA_CERT_REMINDER_MINUTE', 30)
|
|
tasks.init_scheduled_tasks(None)
|
|
job = jobs['scan_pl_cert_expiry']
|
|
assert job['hour'] == 9
|
|
assert job['minute'] == 30
|
|
assert job['timezone'] == 'Asia/Shanghai'
|
|
assert job['max_instances'] == 1
|
|
assert job['coalesce'] is True
|
|
|
|
|
|
def test_scan_uses_shanghai_day_even_when_host_is_previous_day(monkeypatch):
|
|
from datetime import datetime, timezone, date
|
|
from app.tasks import pl_cert_expiry as task
|
|
captured = []
|
|
class Clock:
|
|
@staticmethod
|
|
def now(tz):
|
|
return datetime(2026, 10, 26, 17, tzinfo=timezone.utc).astimezone(tz)
|
|
monkeypatch.setattr(task, 'datetime', Clock)
|
|
monkeypatch.setattr(settings, 'WECHAT_OA_ENABLED', False)
|
|
monkeypatch.setattr(task, 'SessionLocal', lambda: SimpleNamespace(
|
|
commit=lambda: None, rollback=lambda: None, close=lambda: None))
|
|
monkeypatch.setattr(task, 'scan_cert_expiry', lambda db, *, today: captured.append(today) or 0)
|
|
task.scan_pl_cert_expiry_job()
|
|
assert captured == [date(2026, 10, 27)]
|
|
|
|
|
|
def test_default_certificate_scan_is_daily_at_ten_shanghai():
|
|
from app.core.config import Settings
|
|
assert Settings.model_fields['WECHAT_OA_CERT_REMINDER_HOUR'].default == 10
|
|
assert Settings.model_fields['WECHAT_OA_CERT_REMINDER_MINUTE'].default == 0
|