shanxia2.0/backend/tests/test_pl_scan_cert_expiry.py

"""证件到期扫描：有效期拉长后应关闭旧 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