342 lines
14 KiB
Python
342 lines
14 KiB
Python
"""Missing parent/master identities: real SQLite persistence, never business MySQL."""
|
|
from datetime import date, datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pymysql
|
|
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.schemas.pl.plan import PlanUpdate, PlanDeliveryBody
|
|
from app.services import pl_plan_stop_service as stops
|
|
|
|
|
|
@pytest.fixture
|
|
def db(monkeypatch):
|
|
def forbidden(*args, **kwargs):
|
|
raise AssertionError('Business MySQL connections are forbidden')
|
|
monkeypatch.setattr(pymysql, 'connect', forbidden)
|
|
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()
|
|
|
|
|
|
@pytest.fixture
|
|
def save(db, monkeypatch):
|
|
from app.api.pl import plan as api
|
|
plan = m.PlPlan(id='p', tenant_no='t', plan_no='P', create_date=date(2026, 10, 27),
|
|
belongs_company='shanxia', plan_type='inbound')
|
|
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: (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: {})
|
|
# Address library is outside this PL identity test; validation and writes remain real.
|
|
monkeypatch.setattr(stops, 'enrich_stop_addresses_from_library', lambda *args: None)
|
|
user = SimpleNamespace(id='u', dept_id='dept', to_tenant_user_dict=lambda: {})
|
|
return lambda **values: api.update_plan('p', PlanUpdate(saveMode='draft', **values), db, user)
|
|
|
|
|
|
def main(db, row_id='pp', **kwargs):
|
|
values = dict(id=row_id, tenant_no='t', plan_id='p', product_id='sku', product_no='SKU',
|
|
product_name='产品', quantity=10, actual_qty=3, create_by='creator',
|
|
create_time=datetime(2025, 1, 2))
|
|
values.update(kwargs)
|
|
row = m.PlPlanProduct(**values)
|
|
db.add(row)
|
|
db.flush()
|
|
return row
|
|
|
|
|
|
def product(**kwargs):
|
|
return dict(productId='sku', productNo='SKU', productName='产品', **kwargs)
|
|
|
|
|
|
def delivery(db, stop_id='d', child_id='dp', customer='c', pp='pp'):
|
|
row = m.PlPlanDelivery(id=stop_id, tenant_no='t', plan_id='p', customer_id=customer,
|
|
customer_name='安庆市山峡化工有限公司', seq=1, create_by='creator', create_time=datetime(2025, 1, 2))
|
|
child = m.PlPlanDeliveryProduct(id=child_id, tenant_no='t', plan_id='p', delivery_id=stop_id,
|
|
plan_product_id=pp, product_id='sku', product_no='SKU', product_name='产品',
|
|
plan_qty=10, create_by='creator', create_time=datetime(2025, 1, 2))
|
|
db.add_all([row, child])
|
|
db.flush()
|
|
return row, child
|
|
|
|
|
|
def load(db):
|
|
db.add(m.PlPlanLoad(id='l', tenant_no='t', plan_id='p', customer_id='internal', customer_name='安庆市山峡化工有限公司'))
|
|
db.flush()
|
|
|
|
|
|
def stop_body(**kwargs):
|
|
values = dict(customerId='c', customerName='安庆市山峡化工有限公司', products=[product(planProductId='pp', planQty=12)])
|
|
values.update(kwargs)
|
|
return values
|
|
|
|
|
|
def direct(db, rows):
|
|
stops.replace_plan_stops(db, 't', 'p', None, [PlanDeliveryBody(**row) for row in rows], 'u', 'dept')
|
|
db.flush()
|
|
|
|
|
|
@pytest.mark.parametrize("archived_qty", [None, 2])
|
|
def test_api_main_missing_id_repeated_save_keeps_identity_and_omitted_business_fields(db, save, archived_qty):
|
|
old = main(db, model='规格', package_spec='袋', inbound_qty=archived_qty)
|
|
for _ in range(2):
|
|
save(products=[product(quantity=12)])
|
|
assert db.query(m.PlPlanProduct).count() == 1
|
|
assert old.del_flag == 0 and old.actual_qty == 3 and old.inbound_qty == archived_qty
|
|
assert old.quantity == 12 and old.package_spec == '袋' and old.model == '规格'
|
|
assert old.create_by == 'creator' and old.create_time == datetime(2025, 1, 2)
|
|
|
|
|
|
def test_api_main_and_delivery_parent_child_missing_ids_repeated_save(db, save):
|
|
old = main(db)
|
|
load(db)
|
|
parent, child = delivery(db)
|
|
child.outbound_qty = 8
|
|
db.flush()
|
|
for _ in range(2):
|
|
save(products=[product(quantity=12)], load={'id': 'l'}, deliveries=[stop_body()])
|
|
assert db.query(m.PlPlanProduct).count() == 1
|
|
assert db.query(m.PlPlanDelivery).count() == 1
|
|
assert db.query(m.PlPlanDeliveryProduct).count() == 1
|
|
assert (old.del_flag, parent.del_flag, child.del_flag) == (0, 0, 0)
|
|
assert child.plan_qty == 12 and child.outbound_qty == 8
|
|
assert child.create_time == datetime(2025, 1, 2)
|
|
|
|
|
|
@pytest.mark.parametrize('fields', [{'model': 'A'}, {'packageSpec': '袋'}])
|
|
def test_main_unique_variant_uses_only_submitted_discriminators(db, save, fields):
|
|
first = main(db, model='A', package_spec='袋')
|
|
second = main(db, 'pp2', model='B', package_spec='桶')
|
|
save(products=[product(**fields), product(id='pp2')])
|
|
assert first.del_flag == second.del_flag == 0
|
|
assert db.query(m.PlPlanProduct).count() == 2
|
|
|
|
|
|
@pytest.mark.parametrize('rows', [[product()], [product(), product()]])
|
|
def test_main_ambiguous_or_reused_candidate_rejects_before_header_write(db, save, rows):
|
|
main(db)
|
|
if len(rows) == 1:
|
|
main(db, 'pp2', model='B')
|
|
with pytest.raises(HTTPException) as exc:
|
|
save(remark='must not write', products=rows)
|
|
assert exc.value.status_code == 422
|
|
assert not db.new and not db.dirty
|
|
|
|
|
|
def test_main_explicit_reserves_old_same_sku_new_row_is_allowed(db, save):
|
|
old = main(db)
|
|
save(products=[product(), product(id='pp')])
|
|
assert old.del_flag == 0 and db.query(m.PlPlanProduct).count() == 2
|
|
|
|
|
|
def test_main_add_remove_omitted_and_empty(db, save):
|
|
old = main(db)
|
|
save(remark='header only')
|
|
assert old.del_flag == 0
|
|
save(products=[dict(productId='new', productNo='NEW', productName='新增')])
|
|
assert old.del_flag == 1
|
|
assert db.query(m.PlPlanProduct).filter_by(del_flag=0).one().product_id == 'new'
|
|
save(products=[])
|
|
assert db.query(m.PlPlanProduct).filter_by(del_flag=0).count() == 0
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['main', 'delivery'])
|
|
@pytest.mark.parametrize('scope', [{'tenant_no': 'other'}, {'plan_id': 'other'}, {'del_flag': 1}])
|
|
def test_invalid_explicit_id_rejects_without_mutation(db, save, kind, scope):
|
|
main(db)
|
|
if kind == 'main':
|
|
main(db, 'bad', **scope)
|
|
data = dict(products=[product(id='bad')])
|
|
else:
|
|
row, child = delivery(db, 'bad', 'badchild')
|
|
for key, value in scope.items():
|
|
setattr(row, key, value)
|
|
db.flush()
|
|
data = dict(deliveries=[stop_body(id='bad')])
|
|
with pytest.raises(HTTPException) as exc:
|
|
save(remark='must not write', **data)
|
|
assert exc.value.status_code == 422
|
|
assert not db.new and not db.dirty
|
|
|
|
|
|
@pytest.mark.parametrize('with_child_ids', [False, True])
|
|
def test_delivery_same_customer_multiple_stops_needs_reliable_child_ids(db, with_child_ids):
|
|
main(db)
|
|
first, first_child = delivery(db)
|
|
second, second_child = delivery(db, 'd2', 'dp2')
|
|
bodies = [stop_body(products=[product(id='dp2', planProductId='pp')])] if with_child_ids else [stop_body(seq=2)]
|
|
if not with_child_ids:
|
|
with pytest.raises(HTTPException) as exc:
|
|
direct(db, bodies)
|
|
assert exc.value.status_code == 422 and not db.new and not db.dirty
|
|
else:
|
|
direct(db, bodies)
|
|
assert second.del_flag == second_child.del_flag == 0
|
|
assert first.del_flag == first_child.del_flag == 1
|
|
assert db.query(m.PlPlanDelivery).count() == 2
|
|
|
|
|
|
def test_delivery_child_id_can_restore_parent_when_customer_omitted(db):
|
|
main(db)
|
|
parent, child = delivery(db)
|
|
direct(db, [dict(products=[product(id='dp', planProductId='pp')], remark='edit')])
|
|
assert parent.del_flag == child.del_flag == 0
|
|
# Child identity resolves the parent; submitted parent scalars still use
|
|
# the original full-form defaults when omitted.
|
|
assert parent.customer_id is None
|
|
assert parent.remark == 'edit'
|
|
|
|
|
|
@pytest.mark.parametrize('rows', [[dict(remark='no identity')], [stop_body(), stop_body()]])
|
|
def test_delivery_missing_key_or_double_claim_rejected(db, rows):
|
|
main(db)
|
|
delivery(db)
|
|
with pytest.raises(HTTPException) as exc:
|
|
direct(db, rows)
|
|
assert exc.value.status_code == 422 and not db.new and not db.dirty
|
|
|
|
|
|
def test_delivery_explicit_reserves_same_customer_for_new_parent(db):
|
|
main(db)
|
|
parent, child = delivery(db)
|
|
direct(db, [stop_body(), stop_body(id='d')])
|
|
assert parent.del_flag == child.del_flag == 0
|
|
assert db.query(m.PlPlanDelivery).count() == 2
|
|
|
|
|
|
def test_delivery_add_remove_omission_and_empty(db):
|
|
main(db)
|
|
parent, child = delivery(db)
|
|
stops.replace_plan_stops(db, 't', 'p', None, None, 'u', 'dept')
|
|
assert parent.del_flag == 0
|
|
direct(db, [stop_body(customerId='new')])
|
|
assert parent.del_flag == child.del_flag == 1
|
|
direct(db, [])
|
|
assert db.query(m.PlPlanDelivery).filter_by(del_flag=0).count() == 0
|
|
|
|
|
|
def test_delivery_reference_blocks_true_removal_but_not_missing_id_update(db):
|
|
main(db)
|
|
parent, child = delivery(db)
|
|
child.outbound_qty = 2
|
|
db.flush()
|
|
direct(db, [stop_body()])
|
|
assert child.del_flag == 0 and child.outbound_qty == 2
|
|
db.commit()
|
|
with pytest.raises(HTTPException) as exc:
|
|
direct(db, [])
|
|
assert exc.value.status_code == 409 and not db.new and not db.dirty
|
|
|
|
|
|
def test_api_new_master_and_parent_use_preflight_ids_and_client_key(db, save, monkeypatch):
|
|
captured = {}
|
|
original = stops.preflight_plan_child_update
|
|
|
|
def capture(*args):
|
|
result = original(*args)
|
|
captured['products'], captured['stops'] = result
|
|
return result
|
|
|
|
monkeypatch.setattr(stops, 'preflight_plan_child_update', capture)
|
|
save(products=[product(clientKey='request-product', quantity=10)],
|
|
load={'customerId': 'internal', 'customerName': '安庆市山峡化工有限公司'},
|
|
deliveries=[stop_body(products=[product(planProductClientKey='request-product', planQty=10)])])
|
|
pp = db.query(m.PlPlanProduct).one()
|
|
parent = db.query(m.PlPlanDelivery).one()
|
|
child = db.query(m.PlPlanDeliveryProduct).one()
|
|
load_row = db.query(m.PlPlanLoad).one()
|
|
assert pp.id == captured['products'][0]
|
|
assert set(captured['stops'].values()) == {parent.id, load_row.id}
|
|
assert child.delivery_id == parent.id and child.plan_product_id == pp.id
|
|
|
|
|
|
@pytest.mark.parametrize('scope', [{'tenant_no': 'other'}, {'plan_id': 'other'}, {'del_flag': 1}])
|
|
def test_anonymous_main_never_reuses_out_of_scope_or_deleted_row(db, save, scope):
|
|
old = main(db, 'foreign', **scope)
|
|
save(products=[product()])
|
|
active = db.query(m.PlPlanProduct).filter_by(tenant_no='t', plan_id='p', del_flag=0).one()
|
|
assert active.id != old.id
|
|
assert old.actual_qty == 3
|
|
|
|
|
|
@pytest.mark.parametrize('scope', [{'tenant_no': 'other'}, {'plan_id': 'other'}, {'del_flag': 1}])
|
|
def test_anonymous_delivery_child_identity_never_crosses_scope(db, scope):
|
|
main(db)
|
|
parent, child = delivery(db)
|
|
for key, value in scope.items():
|
|
setattr(child, key, value)
|
|
db.flush()
|
|
with pytest.raises(HTTPException) as exc:
|
|
direct(db, [stop_body(products=[product(id='dp', planProductId='pp')])])
|
|
assert exc.value.status_code == 422 and not db.new and not db.dirty
|
|
|
|
|
|
def test_delivery_conflicting_children_cannot_merge_parents(db):
|
|
main(db)
|
|
delivery(db)
|
|
delivery(db, 'd2', 'dp2')
|
|
with pytest.raises(HTTPException) as exc:
|
|
direct(db, [stop_body(products=[product(id='dp', planProductId='pp'), product(id='dp2', planProductId='pp')])])
|
|
assert exc.value.status_code == 422 and not db.new and not db.dirty
|
|
|
|
|
|
def test_api_last_ambiguous_delivery_rejects_before_all_changes(db, save):
|
|
old = main(db)
|
|
first, child = delivery(db)
|
|
delivery(db, 'd2', 'dp2')
|
|
with pytest.raises(HTTPException) as exc:
|
|
save(remark='not written', products=[product(quantity=99)],
|
|
deliveries=[stop_body(customerId='new'), stop_body()])
|
|
assert exc.value.status_code == 422 and not db.new and not db.dirty
|
|
assert old.quantity == 10 and first.del_flag == child.del_flag == 0
|
|
|
|
|
|
def test_duplicate_explicit_main_and_delivery_ids_reject(db, save):
|
|
main(db)
|
|
delivery(db)
|
|
for data in [dict(products=[product(id='pp'), product(id='pp')]),
|
|
dict(deliveries=[stop_body(id='d'), stop_body(id='d')])]:
|
|
with pytest.raises(HTTPException) as exc:
|
|
save(remark='not written', **data)
|
|
assert exc.value.status_code == 422 and not db.new and not db.dirty
|
|
|
|
|
|
def test_main_real_removal_keeps_reference_guard(db, save):
|
|
old = main(db, outbound_qty=2)
|
|
with pytest.raises(HTTPException) as exc:
|
|
save(products=[])
|
|
assert exc.value.status_code == 409 and not db.new and not db.dirty
|
|
assert old.del_flag == 0
|
|
|
|
|
|
def test_two_anonymous_sku_variants_match_keys_after_reordering(db, save):
|
|
first = main(db, model='A', package_spec='袋')
|
|
second = main(db, 'pp2', model='B', package_spec='桶')
|
|
save(products=[product(model='B', packageSpec='桶', quantity=20),
|
|
product(model='A', packageSpec='袋', quantity=30)])
|
|
assert db.query(m.PlPlanProduct).count() == 2
|
|
assert first.del_flag == second.del_flag == 0
|
|
assert first.quantity == 30 and second.quantity == 20
|
|
|
|
|
|
def test_unique_delivery_customer_can_edit_sequence_without_new_identity(db):
|
|
main(db)
|
|
parent, child = delivery(db)
|
|
direct(db, [stop_body(seq=9, address='新地址')])
|
|
assert parent.id == 'd' and parent.seq == 9 and parent.address == '新地址'
|
|
assert parent.del_flag == child.del_flag == 0
|
|
assert db.query(m.PlPlanDelivery).count() == 1
|