274 lines
14 KiB
Python
274 lines
14 KiB
Python
"""SQLite coverage of real stop save/sync; never connect to business databases."""
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
|
|
import pymysql
|
|
import pytest
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.crm_models import CrmProduct
|
|
from app.models.db_models import DbPurchasePlan, DbInboundOutbound, DbWeighingSlip, DbSequence
|
|
from app.models.pl_models import (PlPlan, PlPlanProduct, PlTransport, PlTransportDbLink,
|
|
PlPlanLoad, PlPlanDelivery, PlPlanLoadProduct, PlPlanDeliveryProduct)
|
|
from app.schemas.pl.plan import PlanLoadBody
|
|
from app.services.pl_plan_stop_service import replace_plan_stops
|
|
from app.services import pl_dibang_bridge_service as bridge
|
|
|
|
INTERNAL = '安庆市山峡化工有限公司'
|
|
|
|
@pytest.fixture
|
|
def db(monkeypatch):
|
|
def forbidden(*a, **k):
|
|
raise AssertionError('Business database connection forbidden')
|
|
monkeypatch.setattr(pymysql, 'connect', forbidden)
|
|
monkeypatch.setattr(pymysql.connections.Connection, 'connect', forbidden)
|
|
engine = create_engine('sqlite://')
|
|
for model in (CrmProduct, PlPlan, PlPlanProduct, PlTransport, PlTransportDbLink,
|
|
PlPlanLoad, PlPlanDelivery, PlPlanLoadProduct, PlPlanDeliveryProduct,
|
|
DbPurchasePlan, DbInboundOutbound, DbWeighingSlip, DbSequence):
|
|
model.__table__.create(engine)
|
|
with Session(engine, autoflush=False, expire_on_commit=False) as session:
|
|
session.add(DbSequence(id='seq', tenant_no='t', seq_name='inbound_outbound', next_val=1))
|
|
session.commit()
|
|
yield session
|
|
engine.dispose()
|
|
|
|
|
|
def seed(db, quantities=(5,), pending=False):
|
|
plan = PlPlan(id='p', tenant_no='t', plan_no='P', plan_type='outbound',
|
|
create_date=date(2026, 10, 1), plan_date=date(2026, 10, 1), belongs_company='shanxia',
|
|
transport_id='tr', license_plate='Plate')
|
|
pp = PlPlanProduct(id='pp', tenant_no='t', plan_id='p', product_id='sku',
|
|
product_no='SKU', product_name='Product', model='Spec', unit='吨', quantity=sum(quantities))
|
|
tr = PlTransport(id='tr', tenant_no='t', transport_no='TR', license_plate='Plate', del_flag=0)
|
|
db.add_all([plan, pp, tr, CrmProduct(id='sku', tenant_no='t', product_no='SKU',
|
|
product_name='Product', model='Spec', status='on')]); db.commit()
|
|
load = PlPlanLoad(id='load', tenant_no='t', plan_id='p', customer_name=INTERNAL,
|
|
generate_dibang=0, dibang_direction='outbound', del_flag=0)
|
|
db.add(load)
|
|
for i, qty in enumerate(quantities):
|
|
db.add(PlPlanLoadProduct(id=f'lp{i}', tenant_no='t', plan_id='p', load_id='load',
|
|
plan_product_id='pp', product_id='sku', product_no='SKU', product_name='Product',
|
|
model='Spec', unit='吨', package_spec=str(i), plan_qty=qty, del_flag=0))
|
|
if not pending: db.commit()
|
|
return plan, tr, load
|
|
|
|
|
|
def sync(db, rows, options=None):
|
|
plan, tr, load = rows
|
|
return bridge.sync_dibang_for_plan_stops(db, tenant_no='t', plan=plan, transport=tr,
|
|
stop_options=options if options is not None else [bridge.StopDibangOption('load', load.id, True)],
|
|
user_id='editor', dept_id='dept')
|
|
|
|
|
|
def record(db):
|
|
return (db.query(PlTransportDbLink).filter_by(del_flag=0).one(),
|
|
db.query(DbInboundOutbound).filter_by(del_flag=0).one(), db.query(DbPurchasePlan).one())
|
|
|
|
|
|
def test_same_plan_product_rows_aggregate_and_repeat_keeps_ids_feedback_and_quantity(db):
|
|
rows = seed(db)
|
|
pp = db.get(PlPlanProduct, 'pp')
|
|
body = PlanLoadBody(customerName=INTERNAL, generateDibang=True, products=[
|
|
dict(productId='sku', productNo='SKU', productName='Product', model='Spec', unit='吨',
|
|
id='lp0' if package == '桶' else None, planProductId='pp', planQty=qty, packageSpec=package) for qty, package in ((5,'桶'),(3,'袋'))])
|
|
replace_plan_stops(db, 't', 'p', body, None, 'editor', 'dept', [pp])
|
|
# Deliberately no flush: both real callers can have pending child edits.
|
|
sync(db, rows)
|
|
link, inv, monthly = record(db)
|
|
identity = (link.id, inv.id, link.create_time, inv.create_time)
|
|
inv.site_feedback_remark = 'retain'; inv.site_access_token = 'retain-token'; db.commit()
|
|
sync(db, rows); db.commit()
|
|
assert db.query(PlTransportDbLink).count() == db.query(DbInboundOutbound).count() == 1
|
|
assert (link.id, inv.id, link.create_time, inv.create_time) == identity
|
|
assert link.contributed_qty == inv.quantity == monthly.planned_quantity == 8
|
|
assert (inv.site_feedback_remark, inv.site_access_token) == ('retain', 'retain-token')
|
|
|
|
|
|
@pytest.mark.parametrize('bad', ['deleted_inventory', 'missing_inventory', 'duplicate_link',
|
|
'foreign_inventory', 'missing_month', 'deleted_month', 'wrong_month', 'wrong_link_plan'])
|
|
def test_invalid_link_history_rejected_without_any_writes_or_rebuild(db, bad):
|
|
rows = seed(db); sync(db, rows); db.commit(); link, inv, month = record(db)
|
|
if bad == 'deleted_inventory': inv.del_flag = 1
|
|
elif bad == 'missing_inventory': link.db_inventory_id = 'missing'
|
|
elif bad == 'duplicate_link': db.add(PlTransportDbLink(id='duplicate', tenant_no='t', transport_id='tr',
|
|
plan_id='p', plan_product_id='pp', stop_type='load', stop_id='load',
|
|
db_inventory_id=inv.id, db_purchase_plan_id=month.id, contributed_qty=5))
|
|
elif bad == 'foreign_inventory': inv.tenant_no = 'other'
|
|
elif bad == 'missing_month': link.db_purchase_plan_id = 'missing'
|
|
elif bad == 'deleted_month': month.del_flag = 1
|
|
elif bad == 'wrong_month': inv.purchase_plan_id = 'other'
|
|
else: link.plan_id = 'other'
|
|
db.commit(); writes=[]
|
|
def capture(conn, cursor, sql, *args):
|
|
if sql.lstrip().upper().startswith(('UPDATE','INSERT','DELETE')): writes.append(sql)
|
|
event.listen(db.bind, 'before_cursor_execute', capture)
|
|
try:
|
|
with pytest.raises(ValueError): sync(db, rows)
|
|
finally: event.remove(db.bind, 'before_cursor_execute', capture)
|
|
assert not writes and not db.dirty and not db.new
|
|
assert link.del_flag == 0 and month.planned_quantity == 5
|
|
|
|
|
|
@pytest.mark.parametrize('field,value', [('product_id','other'), ('model','Different'), ('unit','kg')])
|
|
def test_conflicting_same_key_rows_rejected_before_stop_or_inventory_mutation(db, field, value):
|
|
rows = seed(db, (5,3)); setattr(db.get(PlPlanLoadProduct,'lp1'), field, value); db.commit()
|
|
with pytest.raises(ValueError): sync(db, rows)
|
|
assert not db.dirty and not db.new and rows[2].generate_dibang == 0
|
|
assert db.query(DbInboundOutbound).count() == 0
|
|
|
|
|
|
def test_later_stop_error_does_not_update_first_stop_or_options(db):
|
|
rows = seed(db); sync(db, rows); db.commit(); link, inv, month = record(db)
|
|
db.get(PlPlanLoadProduct,'lp0').plan_qty = 9
|
|
db.add(PlPlanDelivery(id='delivery', tenant_no='t', plan_id='p', seq=1,
|
|
customer_name=INTERNAL, generate_dibang=0, dibang_direction='inbound'))
|
|
db.add(PlPlanDeliveryProduct(id='dp', tenant_no='t', plan_id='p', delivery_id='delivery',
|
|
plan_product_id='pp', product_id='missing-sku', product_no='SKU', product_name='Product', plan_qty=5))
|
|
db.commit()
|
|
with pytest.raises(ValueError): sync(db, rows, [bridge.StopDibangOption('load','load',True),
|
|
bridge.StopDibangOption('delivery','delivery',True)])
|
|
assert inv.quantity == month.planned_quantity == 5 and not db.dirty and not db.new
|
|
assert db.get(PlPlanDelivery,'delivery').generate_dibang == 0
|
|
|
|
|
|
@pytest.mark.parametrize('bad', ['foreign_stop', 'missing_stop', 'duplicate_option', 'foreign_product_parent'])
|
|
def test_scope_and_option_validation(db, bad):
|
|
rows = seed(db)
|
|
options=[bridge.StopDibangOption('load','load',True)]
|
|
if bad=='foreign_stop': rows[2].tenant_no='other'
|
|
elif bad=='missing_stop': options=[bridge.StopDibangOption('load','missing',True)]
|
|
elif bad=='duplicate_option': options*=2
|
|
else: db.get(PlPlanLoadProduct,'lp0').plan_id='other'
|
|
db.commit()
|
|
with pytest.raises(ValueError):sync(db, rows, options)
|
|
assert not db.dirty and not db.new and db.query(DbInboundOutbound).count()==0
|
|
|
|
|
|
def test_pending_new_stop_and_transport_are_supported_without_outer_flush(db):
|
|
rows=seed(db,(5,3),pending=True)
|
|
sync(db,rows); db.commit()
|
|
link,inv,month=record(db)
|
|
assert inv.quantity==month.planned_quantity==8
|
|
|
|
|
|
def test_transport_is_locked_before_first_inventory_write(db):
|
|
from sqlalchemy.dialects import mysql
|
|
rows=seed(db); locks=[]
|
|
def observe(state): locks.append(str(state.statement.compile(dialect=mysql.dialect())))
|
|
event.listen(db,'do_orm_execute',observe)
|
|
try: sync(db,rows)
|
|
finally:event.remove(db,'do_orm_execute',observe)
|
|
assert any('FROM pl_transport ' in sql and 'FOR UPDATE' in sql for sql in locks)
|
|
|
|
|
|
def test_new_transport_pending_without_outer_flush(db):
|
|
rows=seed(db,(5,3))
|
|
new=PlTransport(id='new-tr',tenant_no='t',transport_no='NEW',license_plate='New')
|
|
db.add(new);rows[0].transport_id=new.id
|
|
sync(db,(rows[0],new,rows[2])); db.commit()
|
|
link,inv,month=record(db)
|
|
assert link.transport_id=='new-tr' and inv.quantity==month.planned_quantity==8
|
|
|
|
|
|
def test_explicit_cancel_then_reenable_is_a_new_business_action(db):
|
|
rows=seed(db);sync(db,rows);db.commit();old,inv,month=record(db)
|
|
sync(db,rows,[bridge.StopDibangOption('load','load',False)]);db.commit()
|
|
assert old.del_flag==inv.del_flag==1 and month.planned_quantity==0
|
|
sync(db,rows);db.commit();new,current,month=record(db)
|
|
assert new.id!=old.id and current.id!=inv.id and month.planned_quantity==5
|
|
|
|
|
|
def test_omitted_stops_and_empty_options_do_not_clear_existing_links(db):
|
|
rows=seed(db);sync(db,rows);db.commit();link,inv,month=record(db)
|
|
assert sync(db,rows,[])==[]
|
|
assert link.del_flag==inv.del_flag==0 and month.planned_quantity==5
|
|
|
|
|
|
def test_later_stop_dangling_link_does_not_write_earlier_stop(db):
|
|
rows=seed(db);sync(db,rows);db.commit()
|
|
db.add(PlPlanDelivery(id='delivery',tenant_no='t',plan_id='p',seq=1,customer_name=INTERNAL,
|
|
generate_dibang=1,dibang_direction='inbound'))
|
|
db.add(PlPlanDeliveryProduct(id='dp',tenant_no='t',plan_id='p',delivery_id='delivery',
|
|
plan_product_id='pp',product_id='sku',product_no='SKU',product_name='Product',model='Spec',unit='吨',plan_qty=5))
|
|
db.commit();options=[bridge.StopDibangOption('load','load',True),bridge.StopDibangOption('delivery','delivery',True)]
|
|
sync(db,rows,options);db.commit()
|
|
link=db.query(PlTransportDbLink).filter_by(stop_type='delivery').one();link.db_inventory_id='missing'
|
|
db.get(PlPlanLoadProduct,'lp0').plan_qty=9;db.commit()
|
|
with pytest.raises(ValueError):sync(db,rows,options)
|
|
assert not db.dirty and not db.new
|
|
assert db.query(DbInboundOutbound).filter_by(business_type='出库').one().quantity==5
|
|
|
|
|
|
@pytest.mark.parametrize('bad', ['wrong_stop_type','month_business','inventory_multiple_owners'])
|
|
def test_history_scope_details_are_not_repaired_by_saving(db,bad):
|
|
rows=seed(db);sync(db,rows);db.commit();link,inv,month=record(db)
|
|
if bad=='wrong_stop_type':link.stop_type='delivery'
|
|
elif bad=='month_business':month.product_id='other-product'
|
|
else:db.add(PlTransportDbLink(id='other-owner',tenant_no='t',transport_id='other-tr',plan_id='other-plan',
|
|
plan_product_id='other-pp',stop_type='load',stop_id='other-stop',db_inventory_id=inv.id,
|
|
db_purchase_plan_id=month.id,contributed_qty=0))
|
|
db.commit()
|
|
with pytest.raises(ValueError):sync(db,rows)
|
|
assert not db.dirty and not db.new and month.planned_quantity==5
|
|
|
|
|
|
def test_all_stop_link_and_month_existence_reads_are_current_reads(db):
|
|
from sqlalchemy.dialects import mysql
|
|
rows=seed(db);sql=[]
|
|
def observe(state):
|
|
sql.append(str(state.statement.compile(dialect=mysql.dialect())))
|
|
event.listen(db,'do_orm_execute',observe)
|
|
try:
|
|
sync(db,rows);db.commit();sync(db,rows);db.commit()
|
|
sync(db,rows,[bridge.StopDibangOption('load','load',False)])
|
|
finally:event.remove(db,'do_orm_execute',observe)
|
|
# Sequence / human-readable batch-number reads are unrelated to identity decisions.
|
|
identity_reads=[s for s in sql if 'FROM pl_transport_db_link ' in s or 'FROM db_purchase_plan ' in s]
|
|
assert identity_reads and all('FOR UPDATE' in s for s in identity_reads)
|
|
|
|
|
|
def test_pending_month_contribution_is_not_overwritten_by_preflight_refresh(db):
|
|
rows=seed(db);sync(db,rows);db.commit();link,inv,month=record(db)
|
|
month.planned_quantity=12 # same-transaction date synchronizer/other contribution
|
|
sync(db,rows)
|
|
assert month.planned_quantity==12 and inv.quantity==5
|
|
|
|
|
|
def test_zero_quantity_updates_existing_contribution_without_replacing_ids(db):
|
|
rows=seed(db,(5,3));sync(db,rows);db.commit();link,inv,month=record(db)
|
|
identity=(link.id,inv.id)
|
|
for row in db.query(PlPlanLoadProduct):row.plan_qty=0
|
|
sync(db,rows);db.commit()
|
|
assert (link.id,inv.id)==identity and link.del_flag==inv.del_flag==0
|
|
assert link.contributed_qty==inv.quantity==month.planned_quantity==0
|
|
sync(db,rows)
|
|
assert db.query(PlTransportDbLink).count()==db.query(DbInboundOutbound).count()==1
|
|
assert month.planned_quantity==0
|
|
|
|
|
|
def test_first_zero_quantity_does_not_create_empty_link_or_inventory(db):
|
|
rows=seed(db,(0,));sync(db,rows)
|
|
assert db.query(PlTransportDbLink).count()==db.query(DbInboundOutbound).count()==0
|
|
|
|
|
|
def test_cancel_checks_weighing_with_current_tenant_scoped_read(db):
|
|
from sqlalchemy.dialects import mysql
|
|
rows=seed(db);sync(db,rows);db.commit();sql=[]
|
|
def observe(state):sql.append(str(state.statement.compile(dialect=mysql.dialect())))
|
|
event.listen(db,'do_orm_execute',observe)
|
|
try:sync(db,rows,[bridge.StopDibangOption('load','load',False)])
|
|
finally:event.remove(db,'do_orm_execute',observe)
|
|
slips=[s for s in sql if 'FROM db_weighing_slip ' in s]
|
|
assert slips and all('FOR UPDATE' in s and 'db_weighing_slip.tenant_no =' in s for s in slips)
|
|
|
|
|
|
def test_refreshed_owner_cannot_sync_an_old_transport(db):
|
|
from sqlalchemy import text
|
|
rows=seed(db)
|
|
db.execute(text("UPDATE pl_plan SET transport_id='newer-transport' WHERE id='p'"));db.commit()
|
|
assert rows[0].transport_id=='tr'
|
|
with pytest.raises(ValueError):sync(db,rows)
|
|
assert not db.dirty and not db.new
|