shanxia2.0/backend/tests/test_pl_dibang_upsert_on_arrange.py

"""按段 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