shanxia2.0/backend/tests/test_pl_child_differential_save.py

"""Real ORM regressions; exclusively SQLite, never the configured business database."""
import json
from datetime import date
from decimal import Decimal

import pytest
from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from app.models import pl_models as m
from app.models.pl_mobile_stop_meta import PlPlanStopProductMeta
from app.schemas.pl.plan import PlanLoadBody, PlanDeliveryBody, PlanStopProductBody, PlanContractLink
from app.services.pl_plan_stop_service import replace_plan_stops
from app.services.pl_plan_service import replace_plan_contracts
from app.services.pl_brand_archive import persist_plan_product_brand_qtys
from app.services.pl_feedback_brand_qty import persist_transport_product_feedback
from app.services.pl_personnel_service import save_vehicle_personnel
from app.schemas.pl.base_data import VehiclePersonnelRow


@pytest.fixture
def db():
    engine = create_engine('sqlite://')
    for table in m.PlPlan.metadata.sorted_tables:
        if table.name.startswith('pl_'):
            table.create(engine, checkfirst=True)
    with Session(engine, autoflush=False) as session:
        yield session
    engine.dispose()


def seed(db):
    pp=m.PlPlanProduct(id='pp',tenant_no='t',plan_id='p',product_id='sku',product_no='SKU',product_name='产品')
    load=m.PlPlanLoad(id='l',tenant_no='t',plan_id='p',customer_name='安庆山峡')
    delivery=m.PlPlanDelivery(id='d',tenant_no='t',plan_id='p',customer_name='安庆山峡')
    lp=m.PlPlanLoadProduct(id='lp',tenant_no='t',plan_id='p',load_id='l',plan_product_id='pp',product_id='sku',product_no='SKU',product_name='产品',plan_qty=10,inbound_qty=8,archive_remark='保留归档')
    dp=m.PlPlanDeliveryProduct(id='dp',tenant_no='t',plan_id='p',delivery_id='d',plan_product_id='pp',product_id='sku',product_no='SKU',product_name='产品',plan_qty=10)
    db.add_all([pp,load,delivery,lp,dp]);db.flush()
    return pp,lp,dp


def body(row_id):
    return PlanStopProductBody(id=row_id,planProductId='pp',productId='sku',productNo='SKU',productName='产品',planQty=12)


def save_stops(db,pp,load=None,deliveries=None):
    replace_plan_stops(db,'t','p',load,deliveries,'u','dept',[pp])


def test_stop_same_ids_keep_archive_and_creation(db):
    pp,lp,dp=seed(db);created=lp.create_time
    save_stops(db,pp,PlanLoadBody(id='l',products=[body('lp')]),[PlanDeliveryBody(id='d',products=[body('dp')])])
    db.flush();db.expire_all()
    assert db.get(m.PlPlanLoadProduct,'lp').del_flag==0
    assert db.get(m.PlPlanLoadProduct,'lp').inbound_qty==Decimal('8')
    assert db.get(m.PlPlanLoadProduct,'lp').archive_remark=='保留归档'
    assert db.get(m.PlPlanLoadProduct,'lp').create_time==created
    assert db.get(m.PlPlanLoadProduct,'lp').plan_qty==12
    assert db.query(m.PlPlanDeliveryProduct).count()==1


def test_omitted_stop_collections_are_unchanged(db):
    pp,lp,dp=seed(db)
    save_stops(db,pp,PlanLoadBody(id='l',customer_name='更名'),None)
    db.flush();db.expire_all()
    assert db.get(m.PlPlanLoadProduct,'lp').del_flag==0
    assert db.get(m.PlPlanDeliveryProduct,'dp').del_flag==0


@pytest.mark.parametrize('bad_ids',[['foreign'],['dp','dp'],['lp']])
def test_invalid_duplicate_or_wrong_stop_id_rejected_before_mutation(db,bad_ids):
    pp,lp,dp=seed(db)
    with pytest.raises(HTTPException) as exc:
        save_stops(db,pp,PlanLoadBody(id='l',products=[body('lp')]),[PlanDeliveryBody(id='d',products=[body(i) for i in bad_ids])])
    assert exc.value.status_code==422
    assert not db.dirty and not db.new


@pytest.mark.parametrize('reference',['brand','driver','site','snapshot','archive'])
def test_stop_product_removal_rejects_references(db,reference):
    pp,lp,dp=seed(db)
    if reference=='brand':db.add(m.PlPlanProductBrandQty(id='b',tenant_no='t',plan_id='p',plan_product_id='pp',stop_type='delivery',stop_product_id='dp',brand_id='brand',qty=2))
    elif reference in ('driver','site'):
        model=m.PlTransportDriverStopFeedback if reference=='driver' else m.PlTransportSiteStopFeedback
        db.add(model(id='f',tenant_no='t',plan_id='p',transport_id='tr',stop_type='delivery',stop_id='d',product_feedbacks=json.dumps([{'stopProductId':'dp','actualQty':2}]),**({'business_type':'outbound'} if reference=='site' else {})))
    elif reference=='snapshot':db.add(PlPlanStopProductMeta(id='s',tenant_no='t',plan_id='p',stop_type='delivery',stop_id='d',stop_product_id='dp',product_id='sku'))
    else:dp.outbound_qty=2
    db.flush()
    with pytest.raises(HTTPException) as exc:
        save_stops(db,pp,None,[PlanDeliveryBody(id='d',products=[])])
    assert exc.value.status_code==409
    assert dp.del_flag==0


def test_unreferenced_stop_product_deletes_only_removed_row(db):
    pp,lp,dp=seed(db)
    save_stops(db,pp,None,[PlanDeliveryBody(id='d',products=[])])
    db.flush()
    assert dp.del_flag==1 and lp.del_flag==0


def test_contract_diff_keeps_other_sides_and_identity(db):
    old=m.PlPlanContract(id='c',tenant_no='t',plan_id='p',side='supplier',contract_id='contract')
    other=m.PlPlanContract(id='s',tenant_no='t',plan_id='p',side='load',stop_id='l',contract_id='stopcontract')
    db.add_all([old,other]);db.flush()
    replace_plan_contracts(db,tenant_no='t',plan_id='p',supplier_contracts=[PlanContractLink(contractId='contract')],demander_contracts=None,user_id='u',dept_id='dept')
    db.flush();db.expire_all()
    assert db.get(m.PlPlanContract,'c').del_flag==0
    assert db.get(m.PlPlanContract,'s').del_flag==0
    assert db.query(m.PlPlanContract).count()==2


def test_brand_quantity_update_preserves_id(db):
    old=m.PlPlanProductBrandQty(id='b',tenant_no='t',plan_id='p',plan_product_id='pp',brand_id='brand',qty=2)
    db.add(old);db.flush()
    persist_plan_product_brand_qtys(db,tenant_no='t',plan_id='p',plan_product_id='pp',brand_mode='any',rows=[{'brandId':'brand','qty':3}],user_id='u')
    db.flush();db.expire_all()
    assert db.get(m.PlPlanProductBrandQty,'b').del_flag==0
    assert db.get(m.PlPlanProductBrandQty,'b').qty==3
    assert db.query(m.PlPlanProductBrandQty).count()==1


def test_transport_feedback_and_brand_keep_both_ids(db):
    pp,_,_=seed(db);pp.brand_mode='any'
    tr=m.PlTransport(id='tr',tenant_no='t',transport_no='TR');db.add(tr)
    db.add(m.PlTransportProductFeedback(id='f',tenant_no='t',transport_id='tr',plan_id='p',plan_product_id='pp',actual_qty=2))
    db.add(m.PlTransportProductBrandQty(id='b',tenant_no='t',transport_id='tr',feedback_id='f',plan_product_id='pp',brand_id='brand',qty=2));db.flush()
    persist_transport_product_feedback(db,tenant_no='t',transport=tr,plan_id='p',products=[{'planProductId':'pp','actualQty':3,'brandQtys':[{'brandId':'brand','qty':3}]}])
    db.flush();db.expire_all()
    assert db.get(m.PlTransportProductFeedback,'f').del_flag==0
    assert db.get(m.PlTransportProductBrandQty,'b').del_flag==0
    assert db.get(m.PlTransportProductBrandQty,'b').feedback_id=='f'
    assert db.query(m.PlTransportProductFeedback).count()==1


def test_vehicle_personnel_link_keeps_id(db):
    db.add(m.PlPersonnel(id='person',tenant_no='t',personnel_name='人员'))
    db.add(m.PlVehiclePersonnel(id='link',tenant_no='t',vehicle_id='v',personnel_id='person',role='driver'));db.flush()
    save_vehicle_personnel(db,'v',[VehiclePersonnelRow(personnelId='person',personnelName='人员',role='driver')],'t',user_id='u',dept_id='dept')
    db.flush();db.expire_all()
    assert db.get(m.PlVehiclePersonnel,'link').del_flag==0
    assert db.query(m.PlVehiclePersonnel).count()==1


@pytest.mark.parametrize('kind',['inbound','outbound','feedback','dibang'])
def test_main_product_preflight_blocks_downstream_removal(db,kind):
    from app.services import pl_plan_stop_service as service
    from app.schemas.pl.plan import PlanUpdate
    pp,lp,dp=seed(db)
    model={'inbound':m.PlInboundOrderItem,'outbound':m.PlOutboundOrderItem,
           'feedback':m.PlTransportProductFeedback,'dibang':m.PlTransportDbLink}[kind]
    fields=dict(id='ref',tenant_no='t',plan_product_id='pp')
    if kind in ('inbound','outbound'):fields.update(order_id='order',product_id='sku',product_name='产品')
    else:fields.update(plan_id='p',transport_id='tr')
    if kind=='dibang':fields.update(db_inventory_id='inv',db_purchase_plan_id='month')
    db.add(model(**fields));db.flush()
    with pytest.raises(HTTPException) as exc:
        service.preflight_plan_child_update(db,'t','p',PlanUpdate(products=[]))
    assert exc.value.status_code==409
    assert not db.dirty and not db.new


def test_new_product_prefilled_amounts_are_not_archive_evidence(db):
    from app.services.pl_child_references import assert_plan_products_removable
    pp=m.PlPlanProduct(id='pp',tenant_no='t',plan_id='p',product_id='sku',product_no='SKU',product_name='产品',inbound_amount=120,outbound_amount=120)
    db.add(pp);db.flush()
    assert_plan_products_removable(db,'t','p',[pp])
    assert not db.dirty and not db.new


def test_main_product_preflight_rejects_duplicate_and_foreign_ids(db):
    from app.services import pl_plan_stop_service as service
    from app.schemas.pl.plan import PlanUpdate
    seed(db)
    for ids in (['pp','pp'],['foreign']):
        payload=PlanUpdate(products=[dict(id=i,productId='sku',productNo='SKU',productName='产品') for i in ids])
        with pytest.raises(HTTPException) as exc:
            service.preflight_plan_child_update(db,'t','p',payload)
        assert exc.value.status_code==422
        assert not db.dirty and not db.new


def test_partial_stop_validation_keeps_omitted_products_and_deliveries(db):
    from app.services import pl_plan_stop_service as service
    from app.schemas.pl.plan import PlanUpdate
    seed(db)
    payload=PlanUpdate(load=PlanLoadBody(id='l',notes='新说明'))
    load, deliveries=service.complete_plan_stops_for_validation(db,'t','p',payload.load,payload.deliveries)
    assert [p.id for p in load.products]==['lp']
    assert [p.id for p in deliveries[0].products]==['dp']
    assert 'products' not in payload.load.model_fields_set


def test_duplicate_brand_input_rejected_without_mutation(db):
    old=m.PlPlanProductBrandQty(id='b',tenant_no='t',plan_id='p',plan_product_id='pp',brand_id='brand',qty=2)
    db.add(old);db.flush()
    with pytest.raises(HTTPException) as exc:
        persist_plan_product_brand_qtys(db,tenant_no='t',plan_id='p',plan_product_id='pp',brand_mode='any',rows=[{'brandId':'brand','qty':3},{'brandId':'brand','qty':4}],user_id='u')
    assert exc.value.status_code==422
    assert old.qty==2 and not db.dirty


def test_archive_omitted_brands_preserves_existing_brand_rows(db):
    from app.schemas.pl.plan import PlanArchiveProductItem
    from app.services.pl_plan_service import apply_archive_product_quantities
    pp,_,_=seed(db)
    plan=m.PlPlan(id='p',tenant_no='t',plan_no='P',create_date=date(2026,10,27),plan_type='inbound',belongs_company='shanxia')
    brand=m.PlPlanProductBrandQty(id='b',tenant_no='t',plan_id='p',plan_product_id='pp',brand_id='brand',qty=2)
    db.add_all([plan,brand]);db.flush()
    apply_archive_product_quantities(db,tenant_no='t',plan=plan,items=[PlanArchiveProductItem(id='pp',actualQty=2)],user_id='u',sync_from_io_order=False)
    db.flush()
    assert brand.del_flag==0


def test_admin_feedback_omitted_brands_preserves_existing_brand_rows(db):
    from app.schemas.pl.plan import PlanDriverFeedbackUpdateBody
    from app.services.pl_plan_service import apply_plan_driver_feedback
    pp,_,_=seed(db)
    plan=m.PlPlan(id='p',tenant_no='t',plan_no='P',create_date=date(2026,10,27),plan_type='inbound',belongs_company='shanxia',transport_id='tr')
    db.add_all([plan,m.PlTransport(id='tr',tenant_no='t',transport_no='TR'),m.PlTransportProductFeedback(id='f',tenant_no='t',plan_id='p',transport_id='tr',plan_product_id='pp',actual_qty=2)])
    brand=m.PlTransportProductBrandQty(id='b',tenant_no='t',transport_id='tr',feedback_id='f',plan_product_id='pp',brand_id='brand',qty=2)
    db.add(brand);db.flush()
    apply_plan_driver_feedback(db,tenant_no='t',plan=plan,payload=PlanDriverFeedbackUpdateBody(products=[{'planProductId':'pp','actualQty':2}]),user_id='u')
    db.flush()
    assert brand.del_flag==0


def test_main_product_partial_edit_keeps_brand_and_archive_fields(db):
    from app.schemas.pl.plan import PlanProductCreate
    from app.services.pl_plan_service import apply_plan_product_fields
    pp,_,_=seed(db)
    pp.brand_mode='any';pp.actual_qty=8;pp.inbound_price=10;pp.inbound_amount=80
    pp.model='原规格';db.flush()
    apply_plan_product_fields(pp,plan_type='inbound',prod=PlanProductCreate(id='pp',productId='sku',productNo='SKU',productName='新显示名'),order_num=0,user_id='u')
    assert pp.brand_mode=='any' and pp.model=='原规格'
    assert pp.actual_qty==8 and pp.inbound_price==10 and pp.inbound_amount==80


def test_stop_feedback_omitted_nested_brands_survive_native_quantity_edit(db):
    from app.services.pl_transport_stop_feedback_service import save_driver_stop_feedback
    old=m.PlTransportDriverStopFeedback(id='feedback',tenant_no='t',plan_id='p',transport_id='tr',stop_type='delivery',stop_id='d',product_feedbacks=json.dumps([{'stopProductId':'dp','actualQty':2,'brandQtys':[{'brandId':'b','qty':2}]}]))
    db.add(old);db.flush()
    save_driver_stop_feedback(db,tenant_no='t',transport_id='tr',plan_id='p',stop_type='delivery',stop_id='d',attachment_urls=[],products=[{'stopProductId':'dp','actualQty':3}])
    assert json.loads(old.product_feedbacks)[0]['brandQtys']==[{'brandId':'b','qty':2}]


def test_duplicate_stop_feedback_product_ids_rejected(db):
    from app.services.pl_transport_stop_feedback_service import save_driver_stop_feedback
    with pytest.raises(HTTPException) as exc:
        save_driver_stop_feedback(db,tenant_no='t',transport_id='tr',plan_id='p',stop_type='delivery',stop_id='d',attachment_urls=[],products=[{'stopProductId':'dp','actualQty':2}]*2)
    assert exc.value.status_code==422
    assert not db.new and not db.dirty


def test_stop_product_add_keeps_existing_and_creates_only_new_row(db):
    pp,lp,dp=seed(db)
    save_stops(db,pp,None,[PlanDeliveryBody(id='d',products=[body('dp'),body(None)])])
    db.flush()
    rows=db.query(m.PlPlanDeliveryProduct).filter_by(del_flag=0).all()
    assert len(rows)==2 and {row.id for row in rows}>={'dp'}
    assert lp.del_flag==0


def test_deleted_load_singleton_reused_with_consistent_contract_scope(db):
    old=m.PlPlanLoad(id='deleted-load',tenant_no='t',plan_id='p',del_flag=1)
    pp=m.PlPlanProduct(id='pp',tenant_no='t',plan_id='p',product_id='sku',product_no='SKU',product_name='产品')
    db.add_all([old,pp]);db.flush()
    save_stops(db,pp,PlanLoadBody(products=[body(None)],contracts=[PlanContractLink(contractId='c')]),[])
    db.flush()
    row=db.query(m.PlPlanLoadProduct).one();contract=db.query(m.PlPlanContract).one()
    assert row.load_id==contract.stop_id==old.id


@pytest.mark.parametrize('part',['product','load','delivery','stop_product'])
def test_create_rejects_carried_source_ids(part):
    from app.services import pl_plan_stop_service as service
    from app.schemas.pl.plan import PlanCreateFull
    payload=PlanCreateFull(belongsCompany='shanxia')
    if part=='product':
        from app.schemas.pl.plan import PlanProductCreate
        payload.products=[PlanProductCreate(id='source',productId='sku',productNo='SKU',productName='产品')]
    elif part=='load':payload.load=PlanLoadBody(id='source')
    elif part=='delivery':payload.deliveries=[PlanDeliveryBody(id='source')]
    else:payload.load=PlanLoadBody(products=[body('source')])
    with pytest.raises(HTTPException) as exc:
        service.assert_plan_create_child_ids(payload)
    assert exc.value.status_code==422


@pytest.mark.parametrize('explicit',[True,False])
def test_explicit_plan_product_link_cannot_point_to_another_product(db,explicit):
    from app.services.pl_plan_stop_service import _resolve_plan_product_id
    pp,_,_=seed(db)
    wrong=body(None).model_copy(update={'product_id':'different'})
    if not explicit:
        wrong=wrong.model_copy(update={'plan_product_id':None})
    with pytest.raises(HTTPException) as exc:
        _resolve_plan_product_id(wrong,[pp])
    assert exc.value.status_code==422


def test_legacy_product_link_without_id_must_be_unambiguous(db):
    from app.services.pl_plan_stop_service import _resolve_plan_product_id
    pp,_,_=seed(db)
    another=m.PlPlanProduct(id='pp2',plan_id='p',tenant_no='t',product_id='sku',product_no='SKU',product_name='产品')
    with pytest.raises(HTTPException) as exc:
        _resolve_plan_product_id(body(None).model_copy(update={'plan_product_id':None}),[pp,another])
    assert exc.value.status_code==422


def test_update_api_partial_product_edit_retains_plan_total(db,monkeypatch):
    from types import SimpleNamespace
    from app.api.pl import plan as api
    from app.schemas.pl.plan import PlanUpdate
    pp,_,_=seed(db);pp.tax_subtotal=120
    plan=m.PlPlan(id='p',tenant_no='t',plan_no='P',create_date=date(2026,10,27),belongs_company='shanxia',plan_type='inbound',total_amount=120)
    db.add(plan);db.flush()
    monkeypatch.setattr(api,'get_tenant_no',lambda *args:'t')
    monkeypatch.setattr(api,'_get_plan',lambda *args:plan)
    monkeypatch.setattr(api,'collect_hazard_product_ids',lambda *args:set())
    monkeypatch.setattr(api,'resolve_hazard_flags_from_crm',lambda *args,**kwargs:(0,0))
    monkeypatch.setattr(api,'sync_linked_dibang_dates_for_plan',lambda *args,**kwargs:[])
    monkeypatch.setattr(api,'load_plan_contracts_grouped',lambda *args,**kwargs:{})
    monkeypatch.setattr(api,'load_plan_stops_grouped',lambda *args,**kwargs:{})
    user=SimpleNamespace(id='u',dept_id='d',to_tenant_user_dict=lambda:{})
    result=api.update_plan('p',PlanUpdate(saveMode='draft',products=[{'id':'pp','productId':'sku','productNo':'SKU','productName':'新名'}]),db,user)
    assert result.total_amount==120 and pp.tax_subtotal==120