wuzilongandCursor
bde19c0644
执行中的运输计划不再随意改单；库存流水记到装卸产品明细；手工派车日历区分酸碱并记录数量。同时收紧合同附件与子表的差量保存，并清理已废弃的 bak 备份目录。 Co-authored-by: Cursor <cursoragent@cursor.com>
364 lines
18 KiB
Python
364 lines
18 KiB
Python
"""Real SQLite inventory regressions; business databases are forbidden."""
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
import pymysql
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import Session
|
|
from app.models import pl_models as m
|
|
from app.models.crm_models import CrmProduct
|
|
from app.services.pl_inventory_service import (
|
|
apply_archive_inventory_for_plan, apply_archive_inventory_for_plan_stops,
|
|
reverse_plan_inventory_postings,
|
|
)
|
|
|
|
@pytest.fixture
|
|
def db(monkeypatch):
|
|
def forbidden(*a, **kw):
|
|
raise AssertionError('Business DB forbidden')
|
|
monkeypatch.setattr(pymysql, 'connect', forbidden)
|
|
monkeypatch.setattr(pymysql.connections.Connection, 'connect', forbidden)
|
|
engine = create_engine('sqlite://')
|
|
for model in (CrmProduct, m.PlPlan, m.PlPlanProduct, m.PlPlanProductBrandQty,
|
|
m.PlInventory, m.PlInventoryTransaction, m.PlPlanLoad, m.PlPlanDelivery,
|
|
m.PlPlanLoadProduct, m.PlPlanDeliveryProduct):
|
|
model.__table__.create(engine)
|
|
with Session(engine, autoflush=False, expire_on_commit=False) as session:
|
|
yield session
|
|
engine.dispose()
|
|
|
|
|
|
def seed(db, *, tenant='t', pid='p', qty='10', price='5', direction='inbound', master=True):
|
|
if master and db.get(CrmProduct, 'sku') is None:
|
|
db.add(CrmProduct(id='sku', tenant_no=tenant, product_no='SKU', product_name='product'))
|
|
plan = m.PlPlan(id=pid, tenant_no=tenant, plan_no=pid, create_date=date(2020,1,1),
|
|
belongs_company='shanxia', plan_type=direction, archive_status='archived')
|
|
product = m.PlPlanProduct(id=pid+'-pp', tenant_no=tenant, plan_id=pid,
|
|
product_id='sku', product_no='SKU', product_name='product', actual_qty=Decimal(qty),
|
|
inbound_price=Decimal(price), outbound_price=Decimal(price))
|
|
db.add_all([plan, product]); db.commit()
|
|
return plan, product
|
|
|
|
|
|
def archive(db, plan, *, stops=False, tenant='t'):
|
|
fn = apply_archive_inventory_for_plan_stops if stops else apply_archive_inventory_for_plan
|
|
fn(db, tenant_no=tenant, plan=plan, user_id='editor', dept_id='dept')
|
|
db.flush()
|
|
|
|
|
|
def rows(db, pid='p'):
|
|
return db.query(m.PlInventoryTransaction).filter_by(plan_id=pid, del_flag=0).all()
|
|
|
|
|
|
def inv(db):
|
|
return db.query(m.PlInventory).filter_by(tenant_no='t', product_id='sku', brand_id='', del_flag=0).one()
|
|
|
|
|
|
def test_noop_preserves_identity_creation_period_and_snapshot_without_writes(db):
|
|
plan, product = seed(db); archive(db, plan)
|
|
old = rows(db)[0]; old.create_by='creator'; old.create_time=datetime(2020,1,1); old.stock_after=42
|
|
db.commit()
|
|
before=(old.id,old.create_by,old.create_time,old.update_time,old.stock_after)
|
|
writes=[]
|
|
def record(conn,cursor,sql,*args):
|
|
if sql.lstrip().upper().startswith(('UPDATE','INSERT','DELETE')): writes.append(sql)
|
|
event.listen(db.bind, 'before_cursor_execute', record)
|
|
try: archive(db,plan)
|
|
finally: event.remove(db.bind, 'before_cursor_execute', record)
|
|
now=rows(db)[0]
|
|
assert (now.id,now.create_by,now.create_time,now.update_time,now.stock_after)==before
|
|
assert db.query(m.PlInventoryTransaction).count()==1 and not writes
|
|
|
|
|
|
def test_quantity_price_net_delta_does_not_revalue_unrelated_stock(db):
|
|
plan,product=seed(db);archive(db,plan);db.commit()
|
|
old=rows(db)[0]; identity=old.id
|
|
inv(db).stock_qty+=100;inv(db).stock_amount+=1000;db.commit()
|
|
product.actual_qty=12;product.inbound_price=7
|
|
archive(db,plan)
|
|
assert rows(db)[0].id==identity and old.qty==12 and old.amount==84
|
|
assert inv(db).stock_qty==112 and inv(db).stock_amount==1084
|
|
assert old.stock_after==12
|
|
|
|
|
|
def test_price_only_keeps_stock_after(db):
|
|
plan,product=seed(db);archive(db,plan);db.commit()
|
|
old=rows(db)[0];old.stock_after=123;db.commit();product.inbound_price=8
|
|
archive(db,plan)
|
|
assert rows(db)[0].id==old.id and old.stock_after==123
|
|
assert inv(db).stock_qty==10 and inv(db).stock_amount==80
|
|
|
|
|
|
def test_explicit_zero_price_does_not_use_tax_price(db):
|
|
plan,product=seed(db,price='0');product.tax_price=20
|
|
archive(db,plan)
|
|
assert rows(db)[0].unit_price==0 and rows(db)[0].amount==0 and inv(db).stock_amount==0
|
|
|
|
|
|
def test_remove_and_reverse_do_not_replace_other_plan(db):
|
|
plan,product=seed(db)
|
|
other,_=seed(db,pid='other',qty='4',price='3')
|
|
archive(db,plan);archive(db,other);db.commit()
|
|
removed=rows(db)[0];other_id=rows(db,'other')[0].id
|
|
assert inv(db).stock_amount==62
|
|
product.actual_qty=0;archive(db,plan)
|
|
assert removed.del_flag==1 and rows(db)==[]
|
|
assert inv(db).stock_qty==4 and inv(db).stock_amount==12
|
|
assert rows(db,'other')[0].id==other_id
|
|
for _ in range(2):
|
|
reverse_plan_inventory_postings(db,tenant_no='t',plan_id='other',user_id='u');db.flush()
|
|
assert rows(db,'other')==[] and inv(db).stock_qty==0 and inv(db).stock_amount==0
|
|
|
|
|
|
def delivery(db,plan,product,identity,qty,price):
|
|
db.add(m.PlPlanDelivery(id=identity,tenant_no='t',plan_id=plan.id,
|
|
customer_name='安庆市山峡化工有限公司',seq=1))
|
|
row=m.PlPlanDeliveryProduct(id=identity+'-item',tenant_no='t',plan_id=plan.id,
|
|
delivery_id=identity,plan_product_id=product.id,product_id='sku',product_no='SKU',
|
|
product_name='product',outbound_qty=Decimal(qty),outbound_price=Decimal(price))
|
|
db.add(row);db.flush();return row
|
|
|
|
|
|
def test_multiple_same_direction_stops_keep_separate_lines(db):
|
|
plan,product=seed(db,direction='direct')
|
|
delivery(db,plan,product,'one','2','3')
|
|
second=delivery(db,plan,product,'two','3','7')
|
|
archive(db,plan,stops=True);db.commit()
|
|
by_line={r.stop_product_id:r for r in rows(db)}
|
|
assert set(by_line)=={'one-item','two-item'}
|
|
assert by_line['one-item'].qty==2 and by_line['one-item'].amount==6 and by_line['one-item'].unit_price==3
|
|
assert by_line['two-item'].qty==3 and by_line['two-item'].amount==21 and by_line['two-item'].unit_price==7
|
|
assert inv(db).stock_qty==5 and inv(db).stock_amount==27
|
|
identities={key:(row.id,row.create_time) for key,row in by_line.items()}
|
|
archive(db,plan,stops=True)
|
|
by_line={r.stop_product_id:r for r in rows(db)}
|
|
assert {key:(row.id,row.create_time) for key,row in by_line.items()}==identities
|
|
second.outbound_qty=4;archive(db,plan,stops=True)
|
|
by_line={r.stop_product_id:r for r in rows(db)}
|
|
assert by_line['one-item'].id==identities['one-item'][0] and by_line['one-item'].qty==2
|
|
assert by_line['two-item'].id==identities['two-item'][0] and by_line['two-item'].qty==4
|
|
assert by_line['two-item'].amount==28 and inv(db).stock_qty==6 and inv(db).stock_amount==34
|
|
|
|
|
|
def test_legacy_stop_lines_bind_by_quantity_and_price(db):
|
|
plan,product=seed(db,direction='direct')
|
|
delivery(db,plan,product,'one','2','3')
|
|
second=delivery(db,plan,product,'two','3','7')
|
|
first=m.PlInventoryTransaction(id='legacy-one',tenant_no='t',product_id='sku',brand_id='',
|
|
product_name='product',transaction_type='inbound',plan_id=plan.id,plan_no=plan.plan_no,
|
|
qty=2,unit_price=3,amount=6,stock_after=2,create_time=datetime(2026,9,15))
|
|
other=m.PlInventoryTransaction(id='legacy-two',tenant_no='t',product_id='sku',brand_id='',
|
|
product_name='product',transaction_type='inbound',plan_id=plan.id,plan_no=plan.plan_no,
|
|
qty=3,unit_price=7,amount=21,stock_after=5,create_time=datetime(2026,9,15))
|
|
db.add_all([first,other,m.PlInventory(id='inv',tenant_no='t',product_id='sku',brand_id='',
|
|
product_no='SKU',product_name='product',stock_qty=5,stock_amount=27)]);db.commit()
|
|
second.outbound_qty=4
|
|
archive(db,plan,stops=True)
|
|
by_line={r.stop_product_id:r for r in rows(db)}
|
|
assert by_line['one-item'].id=='legacy-one' and by_line['one-item'].qty==2
|
|
assert by_line['two-item'].id=='legacy-two' and by_line['two-item'].qty==4 and by_line['two-item'].amount==28
|
|
assert by_line['one-item'].create_time==datetime(2026,9,15)
|
|
assert inv(db).stock_qty==6 and inv(db).stock_amount==34
|
|
|
|
|
|
def test_legacy_stop_lines_reject_ambiguous_rematch_without_mutation(db):
|
|
plan,product=seed(db,direction='direct')
|
|
one=delivery(db,plan,product,'one','4','3')
|
|
two=delivery(db,plan,product,'two','6','7')
|
|
first=m.PlInventoryTransaction(id='legacy-one',tenant_no='t',product_id='sku',brand_id='',
|
|
product_name='product',transaction_type='inbound',plan_id=plan.id,plan_no=plan.plan_no,
|
|
qty=2,unit_price=3,amount=6,stock_after=2)
|
|
other=m.PlInventoryTransaction(id='legacy-two',tenant_no='t',product_id='sku',brand_id='',
|
|
product_name='product',transaction_type='inbound',plan_id=plan.id,plan_no=plan.plan_no,
|
|
qty=3,unit_price=7,amount=21,stock_after=5)
|
|
db.add_all([first,other,m.PlInventory(id='inv',tenant_no='t',product_id='sku',brand_id='',
|
|
product_no='SKU',product_name='product',stock_qty=5,stock_amount=27)]);db.commit()
|
|
with pytest.raises(HTTPException) as error: archive(db,plan,stops=True)
|
|
assert error.value.status_code==409
|
|
assert {r.id for r in rows(db)}=={'legacy-one','legacy-two'}
|
|
assert first.qty==2 and other.qty==3 and first.stop_product_id=='' and other.stop_product_id==''
|
|
assert one.outbound_qty==4 and two.outbound_qty==6 and inv(db).stock_qty==5
|
|
assert not db.dirty and not db.new
|
|
|
|
|
|
def test_merged_legacy_stop_line_is_not_split_automatically(db):
|
|
plan,product=seed(db,direction='direct')
|
|
delivery(db,plan,product,'one','2','3')
|
|
delivery(db,plan,product,'two','3','7')
|
|
merged=m.PlInventoryTransaction(id='merged',tenant_no='t',product_id='sku',brand_id='',
|
|
product_name='product',transaction_type='inbound',plan_id=plan.id,plan_no=plan.plan_no,
|
|
qty=5,unit_price=Decimal('5.4000'),amount=27,stock_after=5,create_time=datetime(2026,9,15))
|
|
db.add_all([merged,m.PlInventory(id='inv',tenant_no='t',product_id='sku',brand_id='',
|
|
product_no='SKU',product_name='product',stock_qty=5,stock_amount=27)]);db.commit()
|
|
with pytest.raises(HTTPException) as error: archive(db,plan,stops=True)
|
|
assert error.value.status_code==409 and '合并' in error.value.detail
|
|
assert rows(db)[0].id=='merged' and merged.qty==5 and merged.stop_product_id==''
|
|
assert inv(db).stock_qty==5 and not db.dirty and not db.new
|
|
|
|
|
|
def test_opposite_directions_stay_distinct(db):
|
|
plan,product=seed(db,direction='direct');delivery(db,plan,product,'delivery','5','7')
|
|
db.add(m.PlPlanLoad(id='load',tenant_no='t',plan_id=plan.id,customer_name='安庆市山峡化工有限公司'))
|
|
db.add(m.PlPlanLoadProduct(id='load-item',tenant_no='t',plan_id=plan.id,load_id='load',
|
|
plan_product_id=product.id,product_id='sku',product_no='SKU',product_name='product',
|
|
inbound_qty=3,inbound_price=2));db.flush()
|
|
archive(db,plan,stops=True);db.commit()
|
|
ids={r.transaction_type:r.id for r in rows(db)}
|
|
assert set(ids)=={'inbound','outbound'}
|
|
assert inv(db).stock_qty==2 and inv(db).stock_amount==29
|
|
archive(db,plan,stops=True)
|
|
assert {r.transaction_type:r.id for r in rows(db)}==ids
|
|
|
|
|
|
def test_ambiguous_history_rejected_before_mutation(db):
|
|
plan,product=seed(db);archive(db,plan);db.commit();old=rows(db)[0]
|
|
db.add(m.PlInventoryTransaction(id='duplicate',tenant_no='t',product_id='sku',brand_id='',
|
|
product_name='product',transaction_type='inbound',plan_id='p',plan_no='p',qty=1,
|
|
unit_price=5,amount=5,stock_after=11));db.commit()
|
|
with pytest.raises(HTTPException) as error: archive(db,plan)
|
|
assert error.value.status_code==409
|
|
assert old.del_flag==0 and len(rows(db))==2 and inv(db).stock_qty==10
|
|
assert not db.dirty and not db.new
|
|
|
|
|
|
def test_wrong_tenant_rejected(db):
|
|
plan,product=seed(db,tenant='other')
|
|
with pytest.raises(HTTPException): archive(db,plan)
|
|
assert db.query(m.PlInventory).count()==0
|
|
|
|
|
|
def test_missing_crm_master_does_not_block_historical_plan(db):
|
|
plan,product=seed(db,master=False);archive(db,plan);db.commit();identity=rows(db)[0].id
|
|
archive(db,plan)
|
|
assert rows(db)[0].id==identity
|
|
|
|
|
|
def test_outbound_signed_amount_net_change(db):
|
|
plan,product=seed(db,direction='outbound',qty='3',price='7')
|
|
db.add(m.PlInventory(id='inv',tenant_no='t',product_id='sku',brand_id='',
|
|
product_no='SKU',product_name='product',stock_qty=10,stock_amount=50));db.commit()
|
|
archive(db,plan)
|
|
assert inv(db).stock_qty==7 and inv(db).stock_amount==29
|
|
product.actual_qty=2;archive(db,plan)
|
|
assert inv(db).stock_qty==8 and inv(db).stock_amount==36
|
|
|
|
|
|
def test_rounding_noop_has_no_drift(db):
|
|
plan,product=seed(db,qty='0.333',price='0.1111');archive(db,plan);db.commit();identity=rows(db)[0].id
|
|
for _ in range(3):archive(db,plan);db.commit()
|
|
assert len(rows(db))==1 and rows(db)[0].id==identity
|
|
assert inv(db).stock_amount==Decimal('0.04')
|
|
|
|
|
|
def test_archive_locks_plan_shared_product_and_postings_before_inventory_changes(db):
|
|
from sqlalchemy.dialects import mysql
|
|
plan,product=seed(db)
|
|
statements=[]
|
|
def observe(state):
|
|
statements.append(str(state.statement.compile(dialect=mysql.dialect())))
|
|
event.listen(db, 'do_orm_execute', observe)
|
|
try: archive(db,plan)
|
|
finally: event.remove(db, 'do_orm_execute', observe)
|
|
locks=[sql for sql in statements if 'FOR UPDATE' in sql]
|
|
assert any('FROM pl_plan ' in sql for sql in locks)
|
|
assert any('FROM crm_product ' in sql for sql in locks)
|
|
assert any('FROM pl_inventory_transaction ' in sql for sql in locks)
|
|
assert any('FROM pl_inventory ' in sql for sql in locks)
|
|
|
|
|
|
def test_archive_endpoint_locks_before_mutating_archive_fields():
|
|
import ast
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from app.schemas.pl.plan import PlanArchiveBody
|
|
tree=ast.parse(Path('app/api/pl/plan.py').read_text())
|
|
node=next(n for n in tree.body if isinstance(n,ast.FunctionDef) and n.name=='archive_plan')
|
|
node.decorator_list=[];node.args.defaults=[ast.Constant(None) for _ in node.args.defaults]
|
|
events=[]
|
|
plan=SimpleNamespace(id='p',tenant_no='t',follow_status='completed',archive_status='archived',has_freight=0)
|
|
user=SimpleNamespace(id='u',dept_id='',name='',to_tenant_user_dict=lambda:{})
|
|
ns=dict(user_can_archive_plan=lambda *a:True,get_tenant_no=lambda *a:'t',_get_plan=lambda *a:plan,
|
|
get_plan=lambda *a:plan,record_plan_change=lambda *a,**k:None,
|
|
lock_archive_inventory=lambda *a,**k:events.append('lock'),
|
|
apply_archive_product_quantities=lambda *a,**k:events.append('quantities'),
|
|
apply_archive_inventory_for_plan=lambda *a,**k:events.append('inventory'))
|
|
module=ast.Module(body=[ast.ImportFrom(module='__future__',names=[ast.alias(name='annotations')],level=0),node],type_ignores=[])
|
|
exec(compile(ast.fix_missing_locations(module),'<current archive endpoint>','exec'),ns)
|
|
ns['archive_plan']('p',PlanArchiveBody(products=[{'id':'pp','actualQty':10}]),SimpleNamespace(commit=lambda:None),user)
|
|
assert events==['lock','quantities','inventory']
|
|
|
|
|
|
def test_pending_new_and_removed_brand_rows_are_seen_without_autoflush(db):
|
|
from app.schemas.pl.plan import PlanArchiveProductItem
|
|
from app.services.pl_plan_service import apply_archive_product_quantities
|
|
plan,product=seed(db);product.brand_mode='any';db.commit()
|
|
def update_brands(values):
|
|
apply_archive_product_quantities(db,tenant_no='t',plan=plan,user_id='u',sync_from_io_order=False,
|
|
items=[PlanArchiveProductItem(id=product.id,actualQty=10,brandQtys=values)])
|
|
archive(db,plan)
|
|
update_brands([{'brandId':'b1','qty':4,'unitPrice':2},{'brandId':'b2','qty':6,'unitPrice':3}])
|
|
assert {r.brand_id for r in rows(db)}=={'b1','b2'}
|
|
by_brand={r.brand_id:r for r in rows(db)}
|
|
assert by_brand['b1'].amount==8 and by_brand['b2'].amount==18
|
|
b2_id=by_brand['b2'].id
|
|
update_brands([{'brandId':'b2','qty':10,'unitPrice':4}])
|
|
assert len(rows(db))==1 and rows(db)[0].id==b2_id and rows(db)[0].amount==40
|
|
assert by_brand['b1'].del_flag==1
|
|
|
|
|
|
def test_api_lock_refreshes_plan_state_but_service_lock_preserves_pending_fields(db):
|
|
from sqlalchemy import text
|
|
from app.services.pl_inventory_service import lock_archive_inventory
|
|
plan,product=seed(db)
|
|
db.execute(text("UPDATE pl_plan SET follow_status='cancelled', archive_status='cancelled' WHERE id='p'"))
|
|
db.commit()
|
|
assert plan.follow_status!='cancelled'
|
|
lock_archive_inventory(db,tenant_no='t',plan=plan,refresh_plan=True)
|
|
assert plan.follow_status=='cancelled' and plan.archive_status=='cancelled'
|
|
plan.remark='pending application change'
|
|
lock_archive_inventory(db,tenant_no='t',plan=plan)
|
|
assert plan.remark=='pending application change'
|
|
|
|
|
|
def test_outbound_price_edit_uses_outbound_price_not_stale_inbound_price(db):
|
|
plan,product=seed(db,direction='outbound')
|
|
archive(db,plan);db.commit();identity=rows(db)[0].id
|
|
product.outbound_price=8
|
|
archive(db,plan)
|
|
assert rows(db)[0].id==identity and rows(db)[0].amount==80
|
|
assert inv(db).stock_amount==Decimal('-80')
|
|
|
|
|
|
def test_api_refresh_reads_latest_product_and_stop_fields(db):
|
|
from sqlalchemy import text
|
|
from app.services.pl_inventory_service import lock_archive_inventory
|
|
plan,product=seed(db,direction='direct')
|
|
stop_product=delivery(db,plan,product,'one','2','3');db.commit()
|
|
db.execute(text("UPDATE pl_plan_product SET inbound_price=9 WHERE id='p-pp'"))
|
|
db.execute(text("UPDATE pl_plan_delivery_product SET outbound_qty=7 WHERE id='one-item'"));db.commit()
|
|
assert product.inbound_price==5 and stop_product.outbound_qty==2
|
|
lock_archive_inventory(db,tenant_no='t',plan=plan,refresh_plan=True)
|
|
assert product.inbound_price==9 and stop_product.outbound_qty==7
|
|
|
|
|
|
def test_other_tenant_postings_and_inventory_untouched(db):
|
|
plan,product=seed(db)
|
|
other=m.PlInventoryTransaction(id='foreign',tenant_no='other',product_id='sku',brand_id='',
|
|
product_name='product',transaction_type='inbound',plan_id=plan.id,plan_no='p',
|
|
qty=50,unit_price=2,amount=100,stock_after=50)
|
|
foreign_inv=m.PlInventory(id='foreign-inv',tenant_no='other',product_id='sku',brand_id='',
|
|
product_no='SKU',product_name='product',stock_qty=50,stock_amount=100)
|
|
db.add_all([other,foreign_inv]);db.commit()
|
|
archive(db,plan);archive(db,plan)
|
|
assert other.del_flag==0 and other.qty==50 and other.stock_after==50
|
|
assert foreign_inv.stock_qty==50 and foreign_inv.stock_amount==100
|
|
|
|
|
|
def test_missing_inventory_for_existing_postings_rejected_without_repair(db):
|
|
plan,product=seed(db);archive(db,plan);db.commit();old=rows(db)[0]
|
|
inv(db).del_flag=1;db.commit()
|
|
with pytest.raises(HTTPException) as exc: archive(db,plan)
|
|
assert exc.value.status_code==409
|
|
assert old.del_flag==0 and db.query(m.PlInventory).count()==1
|