shanxia2.0/backend/tests/test_wechat_cert_notices.py

"""证件每日汇总以真实证件、当前角色和现有列表数据权限为准。"""
import importlib
import importlib.util
import json
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace

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


def test_cert_notice_module_exists():
    assert importlib.util.find_spec("app.services.wechat_cert_notices") is not None


@pytest.fixture
def cert_notices(monkeypatch):
    assert importlib.util.find_spec("app.services.wechat_cert_notices") is not None
    svc = importlib.import_module("app.services.wechat_cert_notices")
    from app.services import wechat_notice_events
    from app.models.models import SysRole, SysRoleDataScopeDetail, SysUser, SysUserRole
    from app.models.pl_models import PlCert, PlDocumentReminder
    from app.models.wechat_notice import WechatNotice

    engine = create_engine("sqlite://")
    for model in (SysRole, SysRoleDataScopeDetail, SysUser, SysUserRole,
                  PlCert, PlDocumentReminder, WechatNotice):
        model.__table__.create(engine)
    db = sessionmaker(bind=engine, autoflush=False)()
    monkeypatch.setattr(svc, "settings", SimpleNamespace(
        WECHAT_OA_ENABLED=True, WECHAT_OA_CERT_REMINDER_ENABLED=True,
        WECHAT_OA_CERT_REMINDER_HOUR=10, WECHAT_OA_CERT_REMINDER_MINUTE=0,
        WECHAT_OA_TEMPLATE_CERT_REMINDER="aYzLZ_EqI8hq7RXrlf73BTAfNvzgJHeg_xWKar01Md0",
        WECHAT_OA_CERT_NOTICE_ROLE_NAME="证件到期通知接收人",
    ))
    monkeypatch.setattr(wechat_notice_events.settings, "WECHAT_OA_ENABLED", True)
    for tenant in ("t1", "t2"):
        for role_id, name, scope in (("notice", "证件到期通知接收人", "5"),
                                     ("self", "本人", "4")):
            db.add(SysRole(id=f"{tenant}-{role_id}", tenant_no=tenant,
                           role_name=name, data_scope=scope))
        for uid in ("a", "b"):
            user_id = f"{tenant}-{uid}"
            db.add(SysUser(id=user_id, tenant_no=tenant, user_no=user_id,
                           nick_name=user_id, dept_id=f"{tenant}-dept"))
            for role_id in ("notice", "self"):
                db.add(SysUserRole(id=f"{user_id}-{role_id}", tenant_no=tenant,
                                  user_id=user_id, role_id=f"{tenant}-{role_id}"))
    db.commit()
    now = datetime(2026, 10, 26, 10, 0)

    def add_cert(cid="cert-1", tenant="t1", creator="t1-a", days=10,
                 reminder_creator=None, **kwargs):
        cert = PlCert(id=cid, tenant_no=tenant, owner_type="personnel", owner_id="owner",
                      cert_name="从业资格证", expire_date=now.date() + timedelta(days=days),
                      remind_days=30, create_by=creator, **kwargs)
        db.add(cert)
        db.add(PlDocumentReminder(id=f"reminder-{cid}", tenant_no=tenant, cert_id=cid,
                                 document_type=cert.cert_name, expire_date=cert.expire_date,
                                 days_remaining=days, create_by=reminder_creator or creator))
        db.flush()
        return cert

    yield svc, db, now, add_cert, WechatNotice
    db.close()
    engine.dispose()


@pytest.mark.parametrize("flag", ["WECHAT_OA_ENABLED", "WECHAT_OA_CERT_REMINDER_ENABLED"])
def test_disabled_skips_database(cert_notices, flag):
    svc, _, now, _, _ = cert_notices
    setattr(svc.settings, flag, False)
    assert svc.enqueue_daily_cert_notices(None, now=now) == 0
    assert svc.refresh_cert_notice(None, None, now=now) is False


def test_daily_dedup_template_and_tenant_scope(cert_notices):
    svc, db, now, add_cert, Notice = cert_notices
    add_cert()
    add_cert("cert-2", tenant="t2", creator="t2-b")
    assert svc.enqueue_daily_cert_notices(db, now=now) == 2
    assert svc.enqueue_daily_cert_notices(db, now=now + timedelta(hours=1)) == 0
    rows = db.query(Notice).all()
    assert {(n.tenant_no, n.recipient_id) for n in rows} == {("t1", "t1-a"), ("t2", "t2-b")}
    notice = rows[0]
    assert notice.kind == "cert_daily"
    assert notice.business_id == notice.recipient_id
    assert notice.cycle_id == "2026-10-26"
    assert notice.template_id == svc.settings.WECHAT_OA_TEMPLATE_CERT_REMINDER
    assert json.loads(notice.payload) == {
        "thing5": {"value": "1项即将到期"},
        "time6": {"value": "2026-10-26 10:00:00"},
        "thing3": {"value": "系统默认发出"},
    }


def test_no_commit_and_no_warning_no_send(cert_notices):
    svc, db, now, add_cert, Notice = cert_notices
    add_cert(days=60)
    assert svc.enqueue_daily_cert_notices(db, now=now) == 0
    add_cert("near")
    assert svc.enqueue_daily_cert_notices(db, now=now) == 1
    db.rollback()
    assert db.query(Notice).count() == 0


def test_union_of_roles_and_scope_matches_reminder_list(cert_notices):
    from app.models.models import SysRole, SysRoleDataScopeDetail, SysUserRole
    svc, db, now, add_cert, Notice = cert_notices
    add_cert("by-self")
    add_cert("by-other", creator="other", reminder_creator="other")
    db.add(SysRole(id="custom", tenant_no="t1", role_name="附加可见", data_scope="5"))
    db.add(SysUserRole(id="a-custom", tenant_no="t1", user_id="t1-a", role_id="custom"))
    db.add(SysRoleDataScopeDetail(id="custom-user", tenant_no="t1", role_id="custom",
                                  scope_type="user", scope_value="other"))
    db.flush()
    assert svc.enqueue_daily_cert_notices(db, now=now) == 1
    notice = db.query(Notice).one()
    # 移除本人范围后，另一角色的数据范围仍然有效（并集）。
    db.query(SysUserRole).filter_by(id="t1-a-self").update({"del_flag": 1})
    assert svc.refresh_cert_notice(db, notice, now=now) is True
    db.query(SysRoleDataScopeDetail).filter_by(id="custom-user").update({"del_flag": 1})
    assert svc.refresh_cert_notice(db, notice, now=now) is False


@pytest.mark.parametrize("change", ["renew", "deleted", "revoked", "no_expiry", "resolved",
                                    "user_disabled", "role_disabled", "role_removed", "scope_removed"])
def test_refresh_rechecks_current_business_and_recipient(cert_notices, change):
    from app.models.models import SysRole, SysUser, SysUserRole
    from app.models.pl_models import PlDocumentReminder
    svc, db, now, add_cert, Notice = cert_notices
    cert = add_cert()
    svc.enqueue_daily_cert_notices(db, now=now)
    notice = db.query(Notice).one()
    if change == "renew":
        cert.expire_date = now.date() + timedelta(days=90)
    elif change == "deleted":
        cert.del_flag = 1
    elif change == "revoked":
        cert.status = "revoked"
    elif change == "no_expiry":
        cert.expire_date = None
    elif change == "resolved":
        db.query(PlDocumentReminder).update({"status": "resolved"})
    elif change == "user_disabled":
        db.query(SysUser).filter_by(id="t1-a").update({"status": "0"})
    elif change == "role_disabled":
        db.query(SysRole).filter_by(id="t1-notice").update({"status": "0"})
    else:
        role = "notice" if change == "role_removed" else "self"
        db.query(SysUserRole).filter_by(id=f"t1-a-{role}").update({"del_flag": 1})
    assert svc.refresh_cert_notice(db, notice, now=now) is False


def test_refresh_rebuilds_payload_at_shanghai_send_time_and_expires_next_day(cert_notices):
    svc, db, now, add_cert, Notice = cert_notices
    add_cert()
    svc.enqueue_daily_cert_notices(db, now=now)
    notice = db.query(Notice).one()
    utc_send = datetime(2026, 10, 26, 2, 2, 3, tzinfo=timezone.utc)
    assert svc.refresh_cert_notice(db, notice, now=utc_send) is True
    assert json.loads(notice.payload)["time6"]["value"] == "2026-10-26 10:02:03"
    assert svc.refresh_cert_notice(db, notice, now=datetime(2026, 10, 26, 16, tzinfo=timezone.utc)) is False
    assert svc.enqueue_daily_cert_notices(db, now=now + timedelta(days=2)) == 1


def test_reminder_requires_live_same_tenant_certificate(cert_notices):
    from app.models.pl_models import PlDocumentReminder
    svc, db, now, add_cert, Notice = cert_notices
    add_cert(tenant="t2", creator="t2-a")
    reminder = db.query(PlDocumentReminder).one()
    reminder.tenant_no = "t1"
    reminder.create_by = "t1-a"
    assert svc.enqueue_daily_cert_notices(db, now=now) == 0
    reminder.cert_id = "missing"
    assert svc.enqueue_daily_cert_notices(db, now=now) == 0
    assert db.query(Notice).count() == 0


def test_foreign_tenant_role_assignment_is_ignored(cert_notices):
    from app.models.models import SysUserRole
    svc, db, now, add_cert, _ = cert_notices
    add_cert()
    assignment = db.query(SysUserRole).filter_by(id="t1-a-notice").one()
    assignment.role_id = "t2-notice"
    assert svc.enqueue_daily_cert_notices(db, now=now) == 0


def test_duplicate_reminders_and_stale_remaining_days_do_not_change_daily_gate(cert_notices):
    from app.models.pl_models import PlDocumentReminder
    svc, db, now, add_cert, Notice = cert_notices
    cert = add_cert(days=30)
    reminder = db.query(PlDocumentReminder).one()
    reminder.days_remaining = 900
    db.add(PlDocumentReminder(id="duplicate", tenant_no="t1", cert_id=cert.id,
                             document_type=cert.cert_name, expire_date=cert.expire_date,
                             days_remaining=900, create_by="t1-a"))
    assert svc.enqueue_daily_cert_notices(db, now=now) == 1
    assert db.query(Notice).count() == 1
    assert json.loads(db.query(Notice).one().payload)["thing5"]["value"] == "1项即将到期"
    cert.expire_date = now.date() + timedelta(days=31)
    assert svc.refresh_cert_notice(db, db.query(Notice).one(), now=now) is False


def test_summary_counts_expired_and_upcoming_and_refreshes_after_renewal(cert_notices):
    svc, db, now, add_cert, Notice = cert_notices
    add_cert("upcoming", days=2)
    expired = add_cert("expired", days=-2)
    add_cert("today", days=0)
    add_cert("hidden", creator="other", days=-1)
    svc.enqueue_daily_cert_notices(db, now=now)
    notice = db.query(Notice).one()
    assert json.loads(notice.payload)["thing5"]["value"] == "1项即将到期，2项已到期"
    expired.expire_date = now.date() + timedelta(days=60)
    assert svc.refresh_cert_notice(db, notice, now=now) is True
    assert json.loads(notice.payload)["thing5"]["value"] == "1项即将到期，1项已到期"


def test_long_summary_uses_short_total_title(cert_notices):
    svc, _, now, _, _ = cert_notices
    payload = svc._payload(now, 1234567, 1234567)
    assert payload["thing5"]["value"] == "2469134项证件到期提醒"
    assert len(payload["thing5"]["value"]) <= 20


def test_scope_is_based_on_reminder_creator_as_in_list(cert_notices):
    svc, db, now, add_cert, Notice = cert_notices
    add_cert(creator="t1-a", reminder_creator="t1-b")
    svc.enqueue_daily_cert_notices(db, now=now)
    assert db.query(Notice).one().recipient_id == "t1-b"


@pytest.mark.parametrize("day,hour,expected", [
    (26, 9, 0), (26, 10, 1), (27, 10, 0), (28, 10, 1),
    (29, 10, 0), (30, 10, 1), (31, 10, 0),
])
def test_only_monday_wednesday_friday_after_ten_enqueue(cert_notices, day, hour, expected):
    svc, db, _, add_cert, Notice = cert_notices
    add_cert()
    assert svc.enqueue_daily_cert_notices(db, now=datetime(2026, 10, day, hour)) == expected
    assert db.query(Notice).count() == expected


def test_sunday_and_old_non_schedule_day_notice_never_send(cert_notices):
    svc, db, now, add_cert, Notice = cert_notices
    add_cert()
    svc.enqueue_daily_cert_notices(db, now=now)
    notice = db.query(Notice).one()
    for when in (datetime(2026, 10, 27, 10), datetime(2026, 11, 1, 10),
                 datetime(2026, 10, 26, 9)):
        notice.cycle_id = when.date().isoformat()
        assert not svc.refresh_cert_notice(db, notice, now=when)
    assert svc.enqueue_daily_cert_notices(db, now=datetime(2026, 11, 1, 10)) == 0


def test_enqueue_uses_shanghai_weekday_and_hour(cert_notices):
    svc, db, _, add_cert, Notice = cert_notices
    add_cert()
    # UTC周二晚上是上海周三，但必须等到上海10点。
    assert svc.enqueue_daily_cert_notices(db, now=datetime(2026, 10, 27, 23, tzinfo=timezone.utc)) == 0
    assert svc.enqueue_daily_cert_notices(db, now=datetime(2026, 10, 28, 2, tzinfo=timezone.utc)) == 1
    assert db.query(Notice).one().cycle_id == '2026-10-28'