shanxia2.0/backend/tests/test_pl_plan_copy_seq.py

"""pl_plan.copy_seq_no：租户内 MAX+1，含软删；已有号幂等返回。"""
from datetime import date, datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.models.pl_models import PlPlan
from app.services.pl_driver_plan_service import allocate_or_get_plan_copy_seq

TENANT = "t-copy-seq"


def _session():
    engine = create_engine("sqlite://")
    PlPlan.__table__.create(engine)
    return sessionmaker(bind=engine)()


def _plan(
    *,
    plan_id: str,
    plan_no: str,
    create_time: datetime,
    copy_seq_no=None,
    del_flag: int = 0,
    tenant_no: str = TENANT,
) -> PlPlan:
    return PlPlan(
        id=plan_id,
        plan_no=plan_no,
        copy_seq_no=copy_seq_no,
        create_date=create_time.date() if hasattr(create_time, "date") else date(2026, 1, 1),
        belongs_company="shanxia",
        plan_type="direct",
        follow_status="pending_arrange",
        transport_status="pending",
        archive_status="unarchived",
        complete_status="no",
        has_freight=0,
        total_amount=0,
        freight_amount=0,
        hazard_cert_ready=0,
        erp_entered=0,
        tenant_no=tenant_no,
        create_dept_id="d1",
        create_time=create_time,
        update_time=create_time,
        del_flag=del_flag,
    )


def test_allocate_first_seq_when_tenant_empty():
    db = _session()
    db.add(_plan(plan_id="p1", plan_no="PO1", create_time=datetime(2026, 10, 1, 10, 0, 0)))
    db.commit()

    assert allocate_or_get_plan_copy_seq(db, tenant_no=TENANT, plan_id="p1") == 1
    db.commit()
    assert db.query(PlPlan).get("p1").copy_seq_no == 1


def test_allocate_skips_soft_deleted_max():
    """软删占用的号计入 MAX，不回收。"""
    db = _session()
    db.add(
        _plan(
            plan_id="old",
            plan_no="PO0",
            create_time=datetime(2026, 9, 1, 10, 0, 0),
            copy_seq_no=3,
            del_flag=1,
        )
    )
    db.add(
        _plan(
            plan_id="new",
            plan_no="PO1",
            create_time=datetime(2026, 10, 1, 10, 0, 0),
            copy_seq_no=None,
        )
    )
    db.commit()

    assert allocate_or_get_plan_copy_seq(db, tenant_no=TENANT, plan_id="new") == 4


def test_allocate_idempotent_when_already_set():
    db = _session()
    db.add(
        _plan(
            plan_id="p1",
            plan_no="PO1",
            create_time=datetime(2026, 10, 1, 10, 0, 0),
            copy_seq_no=7,
        )
    )
    db.commit()

    assert allocate_or_get_plan_copy_seq(db, tenant_no=TENANT, plan_id="p1") == 7
    assert db.query(PlPlan).get("p1").copy_seq_no == 7


def test_allocate_missing_plan_returns_none():
    db = _session()
    assert allocate_or_get_plan_copy_seq(db, tenant_no=TENANT, plan_id="missing") is None