81 lines
3.7 KiB
Python
81 lines
3.7 KiB
Python
"""按段 upsert 的原业务策略回归，使用真实 SQLite 身份与历史关联。"""
|
|
from datetime import date, datetime
|
|
import pytest
|
|
from test_pl_stop_dibang_differential import db, seed, sync, record, INTERNAL
|
|
from app.models.db_models import DbInboundOutbound, DbPurchasePlan, DbWeighingSlip
|
|
from app.models.pl_models import PlPlanDelivery
|
|
from app.services.pl_dibang_bridge_service import StopDibangOption
|
|
|
|
|
|
def add_slip(db, inv, month):
|
|
slip = DbWeighingSlip(id='slip', tenant_no='t', inbound_outbound_id=inv.id,
|
|
purchase_plan_id=month.id, vehicle_no='Old', business_type='出库', cargo_name='Product', supplier='S', consignee='C',
|
|
tare_weight=18730, gross_weight=0, net_weight=-18730, gross_weight_time=datetime(2026,10,1), is_voided=0)
|
|
db.add(slip); db.commit()
|
|
return slip
|
|
|
|
|
|
def test_cancel_generate_dibang_blocked_when_weighing_exists(db):
|
|
rows=seed(db); sync(db,rows); db.commit(); link,inv,month=record(db)
|
|
add_slip(db,inv,month)
|
|
with pytest.raises(ValueError,match='已执行|已关联地磅'):
|
|
sync(db,rows,[StopDibangOption('load','load',False)])
|
|
assert link.del_flag==inv.del_flag==0 and month.planned_quantity==5
|
|
assert rows[2].generate_dibang==1
|
|
|
|
|
|
def test_upsert_reuses_existing_inventory_id(db):
|
|
rows=seed(db); first=sync(db,rows); db.commit(); link,inv,month=record(db)
|
|
identity=(link.id,inv.id,inv.batch_no)
|
|
second=sync(db,rows)
|
|
assert first==second and (link.id,inv.id,inv.batch_no)==identity
|
|
assert db.query(DbInboundOutbound).count()==1 and month.planned_quantity==5
|
|
|
|
|
|
def test_upsert_updates_inventory_and_weighing_plate_when_already_weighed(db):
|
|
rows=seed(db); sync(db,rows); db.commit(); link,inv,month=record(db)
|
|
slip=add_slip(db,inv,month); original_id=inv.id
|
|
rows[1].license_plate='New'; rows[2].load_time=datetime(2026,10,6)
|
|
# Caller pending fields must survive the bridge locks.
|
|
results=sync(db,rows)
|
|
assert results[0].db_inventory_id==original_id
|
|
assert inv.vehicle_no==slip.vehicle_no=='New'
|
|
assert slip.tare_weight==18730 and slip.net_weight==-18730
|
|
assert inv.date==date(2026,10,6)
|
|
|
|
|
|
def test_upsert_remounts_monthly_plan_when_partner_changes(db):
|
|
rows=seed(db); sync(db,rows); db.commit(); link,inv,old=record(db)
|
|
original_id=inv.id
|
|
target=DbPurchasePlan(id='target',tenant_no='t',plan_no='SX-NEW',
|
|
plan_date=date(2026,10,1),start_date=date(2026,10,1),end_date=date(2026,10,31),
|
|
business_type='出库',product_id='sku',product='Product',specification='Spec',
|
|
customer_name='New customer',planned_quantity=0)
|
|
db.add(target)
|
|
db.add(PlPlanDelivery(id='delivery',tenant_no='t',plan_id='p',seq=1,
|
|
customer_name='New customer',generate_dibang=0)); db.commit()
|
|
results=sync(db,rows)
|
|
assert len(results)==1 and inv.id==original_id
|
|
assert inv.partner_company=='New customer'
|
|
assert inv.purchase_plan_id==link.db_purchase_plan_id==target.id
|
|
assert inv.plan_no=='SX-NEW' and old.planned_quantity==0 and target.planned_quantity==5
|
|
assert db.query(DbPurchasePlan).count()==2
|
|
|
|
|
|
def test_arrange_transport_source_has_no_revoke_call():
|
|
import inspect
|
|
from app.api.pl import transport as transport_api
|
|
|
|
src = inspect.getsource(transport_api.arrange_transport)
|
|
assert "revoke_regeneratable_dibang_for_transport" not in src
|
|
|
|
|
|
def test_arrange_transport_syncs_stop_dibang_even_when_all_unchecked():
|
|
"""全部取消勾选时仍应进入按段 sync，以便取消/已过磅阻断生效。"""
|
|
import inspect
|
|
from app.api.pl import transport as transport_api
|
|
|
|
src = inspect.getsource(transport_api.arrange_transport)
|
|
assert "bool(stop_dibang_options)" in src
|
|
assert "item.generate_dibang for item in stop_dibang_options" not in src
|