264 lines
12 KiB
Python
264 lines
12 KiB
Python
"""证件每日汇总以真实证件、当前角色和现有列表数据权限为准。"""
|
|
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'
|