wuzilongandCursor
811e6a129f
Add ensure-copy-seq API and miniprogram clipboard prefix so each plan gets a tenant-scoped fixed serial on first copy; cover allocation with unit tests and refresh the backend graph. Co-authored-by: Cursor <cursoragent@cursor.com>
106 lines
2.8 KiB
Python
106 lines
2.8 KiB
Python
"""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
|