298 lines
12 KiB
Python
298 lines
12 KiB
Python
"""Run the legacy bridge against SQLite; never connect to a business database."""
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, update
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.crm_models import CrmProduct
|
|
from app.models.db_models import DbInboundOutbound, DbPurchasePlan, DbSequence, DbWeighingSlip
|
|
from app.models.pl_models import PlPlan, PlPlanProduct, PlTransport, PlTransportDbLink
|
|
from app.services import pl_dibang_bridge_service as bridge
|
|
|
|
|
|
@pytest.fixture
|
|
def db(monkeypatch):
|
|
import pymysql
|
|
monkeypatch.setattr(pymysql, 'connect', lambda *a, **k: pytest.fail('MySQL is forbidden'))
|
|
engine = create_engine('sqlite://')
|
|
for model in (CrmProduct, PlPlan, PlPlanProduct, PlTransport, PlTransportDbLink,
|
|
DbPurchasePlan, DbInboundOutbound, DbWeighingSlip, DbSequence):
|
|
model.__table__.create(engine)
|
|
with Session(engine) as session:
|
|
session.add(DbSequence(id='seq', tenant_no='t', seq_name='inbound_outbound', next_val=1))
|
|
session.commit()
|
|
yield session
|
|
|
|
|
|
def seed(db):
|
|
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),
|
|
pickup_time=datetime(2026, 10, 1), belongs_company='shanxia',
|
|
customer_name='Customer', license_plate='Plate')
|
|
product = PlPlanProduct(id='pp', tenant_no='t', plan_id='p', product_id='sku',
|
|
product_no='SKU', product_name='Product', model='Spec',
|
|
unit='吨', quantity=5)
|
|
transport = PlTransport(id='tr', tenant_no='t', transport_no='TR', license_plate='Plate')
|
|
db.add_all([plan, product, transport, CrmProduct(id='sku', tenant_no='t',
|
|
product_no='SKU', product_name='Product', model='Spec', status='on')])
|
|
db.commit()
|
|
return plan, product, transport
|
|
|
|
|
|
def sync(db, rows, products=None):
|
|
plan, product, transport = rows
|
|
result = bridge.sync_dibang_for_transport(db, tenant_no='t', plan=plan,
|
|
products=products if products is not None else [product], transport=transport,
|
|
user_id='editor', dept_id='dept')
|
|
db.flush()
|
|
return result
|
|
|
|
|
|
def records(db):
|
|
return (db.query(PlTransportDbLink).one(), db.query(DbInboundOutbound).one(),
|
|
db.query(DbPurchasePlan).one())
|
|
|
|
|
|
def test_repeat_reuses_link_inventory_and_monthly_contribution(db):
|
|
rows = seed(db)
|
|
first = sync(db, rows)
|
|
link, inventory, monthly = records(db)
|
|
created = (link.id, inventory.id, link.create_time, inventory.create_time)
|
|
inventory.site_feedback_remark = 'keep feedback'
|
|
inventory.site_feedback_images = '["url"]'
|
|
inventory.site_access_token = 'token'
|
|
db.commit()
|
|
second = sync(db, rows)
|
|
assert db.query(PlTransportDbLink).count() == db.query(DbInboundOutbound).count() == 1
|
|
assert second == first and monthly.planned_quantity == 5
|
|
assert (link.id, inventory.id, link.create_time, inventory.create_time) == created
|
|
assert (inventory.site_feedback_remark, inventory.site_feedback_images,
|
|
inventory.site_access_token) == ('keep feedback', '["url"]', 'token')
|
|
|
|
|
|
def test_quantity_changes_only_apply_delta_and_keep_other_contributions(db):
|
|
rows = seed(db)
|
|
sync(db, rows)
|
|
link, inventory, monthly = records(db)
|
|
monthly.planned_quantity += 20
|
|
rows[1].quantity = 8
|
|
db.commit()
|
|
sync(db, rows)
|
|
assert db.query(DbInboundOutbound).count() == 1
|
|
assert monthly.planned_quantity == 28
|
|
assert inventory.quantity == link.contributed_qty == 8
|
|
rows[1].quantity = 3
|
|
db.commit()
|
|
sync(db, rows)
|
|
assert monthly.planned_quantity == 23 and inventory.quantity == 3
|
|
|
|
|
|
def test_month_migration_moves_only_this_contribution(db):
|
|
rows = seed(db)
|
|
sync(db, rows)
|
|
link, inventory, old = records(db)
|
|
old.planned_quantity += 20
|
|
rows[0].pickup_time = datetime(2026, 11, 2)
|
|
rows[1].quantity = 8
|
|
db.commit()
|
|
sync(db, rows)
|
|
assert db.query(DbInboundOutbound).count() == 1
|
|
new = db.get(DbPurchasePlan, link.db_purchase_plan_id)
|
|
assert new.id != old.id and old.planned_quantity == 20 and new.planned_quantity == 8
|
|
assert inventory.purchase_plan_id == new.id and inventory.date == date(2026, 11, 2)
|
|
sync(db, rows)
|
|
assert new.planned_quantity == 8 and db.query(DbPurchasePlan).count() == 2
|
|
|
|
|
|
def add_weighing(db, inventory):
|
|
slip = DbWeighingSlip(id='weigh', tenant_no='t', inbound_outbound_id=inventory.id,
|
|
vehicle_no='Plate', cargo_name='Product', business_type='出库', supplier='S',
|
|
consignee='C', gross_weight=10000, tare_weight=5200, net_weight=4800,
|
|
gross_weight_time=datetime(2026, 10, 1), is_voided=0)
|
|
inventory.quantity = 4.8
|
|
db.add(slip)
|
|
db.commit()
|
|
return slip
|
|
|
|
|
|
def test_identical_repeat_preserves_actual_weighing_quantity(db):
|
|
rows = seed(db)
|
|
first = sync(db, rows)
|
|
link, inventory, monthly = records(db)
|
|
slip = add_weighing(db, inventory)
|
|
before = (slip.update_time, inventory.update_time)
|
|
assert sync(db, rows) == first
|
|
assert inventory.quantity == 4.8 and monthly.planned_quantity == 5
|
|
assert link.contributed_qty == 5 and slip.net_weight == 4800
|
|
assert (slip.update_time, inventory.update_time) == before
|
|
assert db.query(DbInboundOutbound).count() == 1
|
|
|
|
|
|
@pytest.mark.parametrize('change', ['quantity', 'month', 'customer', 'plate', 'direction'])
|
|
def test_weighed_business_changes_rejected_without_mutation(db, change):
|
|
rows = seed(db)
|
|
sync(db, rows)
|
|
link, inventory, monthly = records(db)
|
|
add_weighing(db, inventory)
|
|
if change == 'quantity': rows[1].quantity = 8
|
|
elif change == 'month': rows[0].pickup_time = datetime(2026, 11, 1)
|
|
elif change == 'customer': rows[0].customer_name = 'Changed'
|
|
elif change == 'plate': rows[2].license_plate = 'Changed'
|
|
else:
|
|
rows[0].plan_type = 'inbound'
|
|
rows[0].supplier_name = 'Customer'
|
|
db.commit()
|
|
with pytest.raises(ValueError):
|
|
sync(db, rows)
|
|
assert not db.dirty and not db.new
|
|
assert inventory.quantity == 4.8 and monthly.planned_quantity == 5
|
|
|
|
|
|
@pytest.mark.parametrize('bad', ['duplicate', 'deleted_link', 'deleted_inventory', 'missing_inventory',
|
|
'foreign_inventory', 'missing_month', 'deleted_month', 'mismatch_month'])
|
|
def test_invalid_history_rejected_instead_of_creating_another_link(db, bad):
|
|
rows = seed(db)
|
|
sync(db, rows)
|
|
link, inventory, monthly = records(db)
|
|
if bad == 'duplicate':
|
|
db.add(PlTransportDbLink(id='duplicate', tenant_no='t', transport_id='tr', plan_id='p',
|
|
plan_product_id='pp', db_inventory_id=inventory.id, db_purchase_plan_id=monthly.id))
|
|
elif bad == 'deleted_link': link.del_flag = 1
|
|
elif bad == 'deleted_inventory': inventory.del_flag = 1
|
|
elif bad == 'missing_inventory': link.db_inventory_id = 'missing'
|
|
elif bad == 'foreign_inventory': inventory.tenant_no = 'other'
|
|
elif bad == 'missing_month': link.db_purchase_plan_id = inventory.purchase_plan_id = 'missing'
|
|
elif bad == 'deleted_month': monthly.del_flag = 1
|
|
else: inventory.purchase_plan_id = 'different'
|
|
db.commit()
|
|
with pytest.raises(ValueError):
|
|
sync(db, rows)
|
|
assert not db.dirty and not db.new
|
|
assert monthly.planned_quantity == 5
|
|
|
|
|
|
def test_multiple_lines_share_one_month_plan_and_validate_all_before_mutation(db):
|
|
rows = seed(db)
|
|
second = PlPlanProduct(id='pp2', tenant_no='t', plan_id='p', product_id='sku',
|
|
product_no='SKU', product_name='Product', model='Spec', unit='吨', quantity=3)
|
|
db.add(second)
|
|
db.commit()
|
|
sync(db, rows, [rows[1], second])
|
|
assert db.query(DbPurchasePlan).count() == 1
|
|
monthly = db.query(DbPurchasePlan).one()
|
|
assert monthly.planned_quantity == 8
|
|
sync(db, rows, [rows[1], second])
|
|
assert monthly.planned_quantity == 8 and db.query(DbInboundOutbound).count() == 2
|
|
rows[1].quantity = 9
|
|
bad_link = db.query(PlTransportDbLink).filter_by(plan_product_id='pp2').one()
|
|
bad_link.db_inventory_id = 'missing'
|
|
db.commit()
|
|
with pytest.raises(ValueError):
|
|
sync(db, rows, [rows[1], second])
|
|
assert monthly.planned_quantity == 8 and not db.dirty and not db.new
|
|
|
|
|
|
def test_duplicate_month_plan_is_rejected_before_new_inventory(db):
|
|
rows = seed(db)
|
|
for identity in ['month1', 'month2']:
|
|
db.add(DbPurchasePlan(id=identity, tenant_no='t', 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='Customer',
|
|
planned_quantity=0))
|
|
db.commit()
|
|
with pytest.raises(ValueError):
|
|
sync(db, rows)
|
|
assert not db.new and not db.dirty and db.query(DbInboundOutbound).count() == 0
|
|
|
|
|
|
@pytest.mark.parametrize('bad', ['duplicate_input', 'foreign_tenant', 'foreign_plan', 'negative'])
|
|
def test_invalid_product_input_rejected_before_new_inventory(db, bad):
|
|
rows = seed(db)
|
|
if bad == 'foreign_tenant': rows[1].tenant_no = 'other'
|
|
elif bad == 'foreign_plan': rows[1].plan_id = 'other'
|
|
elif bad == 'negative': rows[1].quantity = -1
|
|
db.commit()
|
|
with pytest.raises(ValueError):
|
|
sync(db, rows, [rows[1], rows[1]] if bad == 'duplicate_input' else None)
|
|
assert not db.new and not db.dirty and db.query(DbInboundOutbound).count() == 0
|
|
|
|
|
|
def test_pending_direction_change_remounts_without_replacing_inventory(db):
|
|
rows = seed(db)
|
|
sync(db, rows)
|
|
link, inventory, old = records(db)
|
|
rows[0].plan_type = 'inbound'
|
|
rows[0].supplier_name = 'Customer'
|
|
db.commit()
|
|
sync(db, rows)
|
|
new = db.get(DbPurchasePlan, link.db_purchase_plan_id)
|
|
assert new.id != old.id and old.planned_quantity == 0 and new.planned_quantity == 5
|
|
assert inventory.business_type == '入库' and inventory.status == '待入库'
|
|
assert db.query(DbInboundOutbound).count() == 1
|
|
|
|
|
|
def test_unweighed_final_status_still_protects_business_fields(db):
|
|
rows = seed(db)
|
|
sync(db, rows)
|
|
link, inventory, monthly = records(db)
|
|
inventory.status = '已出库'
|
|
db.commit()
|
|
sync(db, rows)
|
|
rows[1].quantity = 8
|
|
db.commit()
|
|
with pytest.raises(ValueError):
|
|
sync(db, rows)
|
|
assert monthly.planned_quantity == 5 and inventory.quantity == 5
|
|
|
|
|
|
def test_missing_contributed_quantity_is_backfilled_only_before_weighing(db):
|
|
rows = seed(db)
|
|
sync(db, rows)
|
|
link, inventory, monthly = records(db)
|
|
link.contributed_qty = None
|
|
db.commit()
|
|
sync(db, rows)
|
|
assert link.contributed_qty == 5 and monthly.planned_quantity == 5
|
|
add_weighing(db, inventory)
|
|
link.contributed_qty = None
|
|
db.commit()
|
|
with pytest.raises(ValueError):
|
|
sync(db, rows)
|
|
assert inventory.quantity == 4.8 and not db.dirty
|
|
|
|
|
|
def test_lock_refreshes_cached_quantities_without_losing_pending_feedback(db):
|
|
rows = seed(db)
|
|
sync(db, rows)
|
|
link, inventory, monthly = records(db)
|
|
# Simulate values newer than this Session's already-loaded identity map.
|
|
db.execute(update(DbPurchasePlan).where(DbPurchasePlan.id == monthly.id)
|
|
.values(planned_quantity=20).execution_options(synchronize_session=False))
|
|
db.execute(update(PlTransportDbLink).where(PlTransportDbLink.id == link.id)
|
|
.values(contributed_qty=4).execution_options(synchronize_session=False))
|
|
assert monthly.planned_quantity == 5 and link.contributed_qty == 5
|
|
inventory.site_feedback_remark = 'pending feedback'
|
|
rows[1].quantity = 6
|
|
sync(db, rows)
|
|
assert monthly.planned_quantity == 22 and link.contributed_qty == 6
|
|
assert inventory.site_feedback_remark == 'pending feedback'
|
|
|
|
|
|
def test_combined_old_contributions_cannot_exceed_month_quantity(db):
|
|
rows = seed(db)
|
|
second = PlPlanProduct(id='pp2', tenant_no='t', plan_id='p', product_id='sku',
|
|
product_no='SKU', product_name='Product', model='Spec', unit='吨', quantity=3)
|
|
db.add(second)
|
|
db.commit()
|
|
sync(db, rows, [rows[1], second])
|
|
monthly = db.query(DbPurchasePlan).one()
|
|
monthly.planned_quantity = 6 # Each old contribution fits separately; their sum does not.
|
|
rows[0].pickup_time = datetime(2026, 11, 1)
|
|
db.commit()
|
|
with pytest.raises(ValueError):
|
|
sync(db, rows, [rows[1], second])
|
|
assert monthly.planned_quantity == 6 and db.query(DbPurchasePlan).count() == 1
|
|
assert not db.dirty and not db.new
|