329 lines
18 KiB
Python
329 lines
18 KiB
Python
"""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
|