shanxia2.0/backend/tests/test_pl_parent_missing_identity.py

"""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