164 lines
4.6 KiB
Python
164 lines
4.6 KiB
Python
"""证件到期扫描：有效期拉长后应关闭旧 pending，待真正进入提醒窗口再生成。"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, timedelta
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.models.pl_models import PlCert, PlDocumentReminder, PlVehicle
|
|
from app.services.pl_cert_service import scan_cert_expiry
|
|
|
|
|
|
def _session():
|
|
engine = create_engine("sqlite://")
|
|
PlCert.__table__.create(engine)
|
|
PlDocumentReminder.__table__.create(engine)
|
|
PlVehicle.__table__.create(engine)
|
|
db = sessionmaker(bind=engine)()
|
|
db.add(
|
|
PlVehicle(
|
|
id="veh-1",
|
|
vehicle_type="self",
|
|
license_plate="皖LB1283",
|
|
tenant_no="t1",
|
|
create_dept_id="d1",
|
|
)
|
|
)
|
|
db.commit()
|
|
return db
|
|
|
|
|
|
def _add_cert(db, *, cert_id: str, expire_date: date, remind_days: int = 30, del_flag: int = 0, status: str = "valid"):
|
|
db.add(
|
|
PlCert(
|
|
id=cert_id,
|
|
owner_type="vehicle",
|
|
owner_id="veh-1",
|
|
cert_name="罐检报告",
|
|
cert_no="皖R8440挂",
|
|
expire_date=expire_date,
|
|
remind_days=remind_days,
|
|
status=status,
|
|
tenant_no="t1",
|
|
create_dept_id="d1",
|
|
del_flag=del_flag,
|
|
)
|
|
)
|
|
|
|
|
|
def _add_pending_reminder(db, *, rem_id: str, cert_id: str, expire_date: date, days_remaining: int):
|
|
db.add(
|
|
PlDocumentReminder(
|
|
id=rem_id,
|
|
cert_id=cert_id,
|
|
owner_type="vehicle",
|
|
owner_id="veh-1",
|
|
owner_label="皖LB1283",
|
|
document_type="罐检报告",
|
|
document_no="皖R8440挂",
|
|
expire_date=expire_date,
|
|
days_remaining=days_remaining,
|
|
status="pending",
|
|
tenant_no="t1",
|
|
create_dept_id="d1",
|
|
)
|
|
)
|
|
|
|
|
|
def test_scan_resolves_pending_when_expire_moved_outside_remind_window():
|
|
"""复现生产：证已改到远期，旧 pending 提醒应被自动关闭。"""
|
|
db = _session()
|
|
today = date.today()
|
|
old_expire = today + timedelta(days=1)
|
|
new_expire = today + timedelta(days=337) # > remind_days=30
|
|
|
|
_add_cert(db, cert_id="cert-tank", expire_date=new_expire, remind_days=30)
|
|
_add_pending_reminder(
|
|
db,
|
|
rem_id="rem-stale",
|
|
cert_id="cert-tank",
|
|
expire_date=old_expire,
|
|
days_remaining=21,
|
|
)
|
|
db.commit()
|
|
|
|
created = scan_cert_expiry(db, tenant_no="t1")
|
|
db.commit()
|
|
|
|
assert created == 0
|
|
rem = db.query(PlDocumentReminder).filter_by(id="rem-stale").one()
|
|
assert rem.status == "resolved"
|
|
assert rem.expire_date == new_expire
|
|
assert rem.days_remaining == (new_expire - today).days
|
|
assert rem.remark and "退出提醒窗口" in rem.remark
|
|
|
|
|
|
def test_scan_keeps_and_updates_pending_inside_remind_window():
|
|
db = _session()
|
|
today = date.today()
|
|
expire = today + timedelta(days=10)
|
|
|
|
_add_cert(db, cert_id="cert-near", expire_date=expire, remind_days=30)
|
|
_add_pending_reminder(
|
|
db,
|
|
rem_id="rem-near",
|
|
cert_id="cert-near",
|
|
expire_date=today + timedelta(days=5),
|
|
days_remaining=5,
|
|
)
|
|
db.commit()
|
|
|
|
created = scan_cert_expiry(db, tenant_no="t1")
|
|
db.commit()
|
|
|
|
assert created == 0
|
|
rem = db.query(PlDocumentReminder).filter_by(id="rem-near").one()
|
|
assert rem.status == "pending"
|
|
assert rem.expire_date == expire
|
|
assert rem.days_remaining == 10
|
|
|
|
|
|
def test_scan_creates_reminder_when_entering_window():
|
|
db = _session()
|
|
today = date.today()
|
|
expire = today + timedelta(days=15)
|
|
|
|
_add_cert(db, cert_id="cert-new", expire_date=expire, remind_days=30)
|
|
db.commit()
|
|
|
|
created = scan_cert_expiry(db, tenant_no="t1")
|
|
db.commit()
|
|
|
|
assert created == 1
|
|
rem = (
|
|
db.query(PlDocumentReminder)
|
|
.filter_by(cert_id="cert-new", status="pending", del_flag=0)
|
|
.one()
|
|
)
|
|
assert rem.expire_date == expire
|
|
assert rem.days_remaining == 15
|
|
|
|
|
|
def test_scan_resolves_pending_when_cert_soft_deleted():
|
|
db = _session()
|
|
today = date.today()
|
|
expire = today + timedelta(days=5)
|
|
|
|
_add_cert(db, cert_id="cert-gone", expire_date=expire, del_flag=1)
|
|
_add_pending_reminder(
|
|
db,
|
|
rem_id="rem-gone",
|
|
cert_id="cert-gone",
|
|
expire_date=expire,
|
|
days_remaining=5,
|
|
)
|
|
db.commit()
|
|
|
|
scan_cert_expiry(db, tenant_no="t1")
|
|
db.commit()
|
|
|
|
rem = db.query(PlDocumentReminder).filter_by(id="rem-gone").one()
|
|
assert rem.status == "resolved"
|
|
assert rem.remark and "失效" in rem.remark
|