shanxia2.0/backend/tests/test_wechat_cert_schedule.py

"""证件预警生成与微信摘要顺序、配置时区和失败隔离。"""
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