shanxia2.0/backend/tests/test_pl_stop_missing_identity.py

"""Legacy stop detail payloads must not silently replace persisted identities."""
from datetime import datetime
from decimal import Decimal

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 PlanDeliveryBody, PlanLoadBody, PlanStopProductBody, PlanUpdate
from app.services.pl_plan_stop_service import preflight_plan_child_update, replace_plan_stops


@pytest.fixture
def db(monkeypatch):
    import pymysql

    def forbidden(*args, **kwargs):
        raise AssertionError('Business database connections are forbidden')

    monkeypatch.setattr(pymysql, 'connect', forbidden)
    monkeypatch.setattr(pymysql.connections.Connection, '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()


def seed(db, kind, *, archive=False):
    pp = m.PlPlanProduct(id='pp', tenant_no='t', plan_id='p', product_id='sku',
                         product_no='SKU', product_name='产品')
    model = m.PlPlanLoadProduct if kind == 'load' else m.PlPlanDeliveryProduct
    stop_model = m.PlPlanLoad if kind == 'load' else m.PlPlanDelivery
    fk = 'load_id' if kind == 'load' else 'delivery_id'
    stop = stop_model(id='stop', tenant_no='t', plan_id='p', customer_name='客户')
    detail = model(id='old', tenant_no='t', plan_id='p', **{fk: 'stop'},
                   plan_product_id='pp', product_id='sku', product_no='SKU',
                   product_name='产品', plan_qty=10, create_by='creator',
                   create_time=datetime(2026, 1, 1), remark='旧备注',
                   **{('inbound_qty' if kind == 'load' else 'outbound_qty'): 8 if archive else None})
    db.add_all([pp, stop, detail])
    db.flush()
    return pp, detail


def product(**values):
    return PlanStopProductBody(**dict(
        {'planProductId': 'pp', 'productId': 'sku', 'productNo': 'SKU',
         'productName': '产品', 'planQty': 12}, **values))


def save(db, pp, kind, products):
    cls = PlanLoadBody if kind == 'load' else PlanDeliveryBody
    stop = cls(id='stop', products=products)
    replace_plan_stops(db, 't', 'p', stop if kind == 'load' else None,
                       [stop] if kind == 'delivery' else None, 'editor', 'dept', [pp])
    db.flush()


@pytest.mark.parametrize('kind', ['load', 'delivery'])
@pytest.mark.parametrize('archive', [False, True])
def test_missing_id_reuses_unique_detail_and_preserves_history(db, kind, archive):
    pp, old = seed(db, kind, archive=archive)
    save(db, pp, kind, [product()])
    db.expire_all()
    assert old.del_flag == 0
    assert old.plan_qty == 12
    assert old.create_time == datetime(2026, 1, 1)
    assert old.create_by == 'creator' and old.remark == '旧备注'
    assert getattr(old, 'inbound_qty' if kind == 'load' else 'outbound_qty') == (Decimal('8') if archive else None)
    assert db.query(type(old)).count() == 1


def test_preflight_accepts_unique_legacy_identity_without_mutation(db):
    pp, old = seed(db, 'load', archive=True)
    payload = PlanUpdate(load=PlanLoadBody(id='stop', products=[product()]))
    preflight_plan_child_update(db, 't', 'p', payload)
    assert old.del_flag == 0 and old.plan_qty == 10
    assert not db.dirty and not db.new


def test_multiple_existing_candidates_require_explicit_ids(db):
    pp, old = seed(db, 'delivery')
    db.add(m.PlPlanDeliveryProduct(id='other', tenant_no='t', plan_id='p', delivery_id='stop',
                                  plan_product_id='pp', product_id='sku', product_no='SKU', product_name='产品'))
    db.flush()
    with pytest.raises(HTTPException) as exc:
        save(db, pp, 'delivery', [product()])
    assert exc.value.status_code == 422
    assert not db.dirty and not db.new


def test_multiple_missing_ids_cannot_compete_for_one_old_row(db):
    pp, old = seed(db, 'delivery')
    with pytest.raises(HTTPException) as exc:
        save(db, pp, 'delivery', [product(), product(planQty=20)])
    assert exc.value.status_code == 422
    assert not db.dirty and not db.new


def test_explicit_ids_reserve_old_rows_before_matching_legacy_rows(db):
    pp, old = seed(db, 'delivery')
    db.add(m.PlPlanDeliveryProduct(id='other', tenant_no='t', plan_id='p', delivery_id='stop',
                                  plan_product_id='pp', product_id='sku', product_no='SKU', product_name='产品'))
    db.flush()
    save(db, pp, 'delivery', [product(), product(id='other', planQty=20)])
    assert old.del_flag == 0 and old.plan_qty == 12
    assert db.get(m.PlPlanDeliveryProduct, 'other').plan_qty == 20
    assert db.query(m.PlPlanDeliveryProduct).count() == 2


def test_explicit_old_row_allows_adding_same_product_as_new_detail(db):
    pp, old = seed(db, 'delivery')
    save(db, pp, 'delivery', [product(), product(id='old')])
    assert old.del_flag == 0
    assert db.query(m.PlPlanDeliveryProduct).filter_by(del_flag=0).count() == 2


def test_contract_key_disambiguates_rows_and_keeps_omitted_contract_fields(db):
    pp, old = seed(db, 'delivery')
    old.contract_id, old.contract_product_id = 'c', 'line-a'
    other = m.PlPlanDeliveryProduct(id='other', tenant_no='t', plan_id='p', delivery_id='stop',
                                   plan_product_id='pp', product_id='sku', product_no='SKU', product_name='产品',
                                   contract_id='c', contract_product_id='line-b')
    db.add(other)
    db.flush()
    save(db, pp, 'delivery', [product(contractProductId='line-a'), product(id='other')])
    assert old.del_flag == 0 and old.contract_id == 'c' and old.contract_product_id == 'line-a'
    assert db.query(m.PlPlanDeliveryProduct).count() == 2


@pytest.mark.parametrize('scope', ['tenant', 'parent', 'stop', 'deleted'])
def test_fallback_never_reuses_out_of_scope_or_deleted_rows(db, scope):
    pp, old = seed(db, 'delivery')
    changes = {'tenant': {'tenant_no': 'other'}, 'parent': {'plan_id': 'other'},
               'stop': {'delivery_id': 'other'}, 'deleted': {'del_flag': 1}}[scope]
    for key, value in changes.items():
        setattr(old, key, value)
    db.flush()
    save(db, pp, 'delivery', [product()])
    assert old.plan_qty == 10
    assert db.query(m.PlPlanDeliveryProduct).count() == 2


def test_unique_legacy_match_and_new_key_insert_keep_existing_identity(db):
    pp, old = seed(db, 'delivery')
    save(db, pp, 'delivery', [product(), product(contractId='new-c', contractProductId='new-line')])
    assert old.del_flag == 0
    assert db.query(m.PlPlanDeliveryProduct).filter_by(del_flag=0).count() == 2


@pytest.mark.parametrize('pending', [False, True])
def test_draft_without_detail_or_product_link_id_preserves_unique_identity(db, pending):
    pp, old = seed(db, 'delivery')
    old.unit = '吨'
    if pending:
        old.plan_product_id = ''
    db.flush()
    replace_plan_stops(db, 't', 'p', None,
                       [PlanDeliveryBody(id='stop', products=[product(planProductId=None)])],
                       'editor', 'dept', [pp], preserve_pending_draft=True)
    db.flush()
    assert old.del_flag == 0 and old.plan_product_id == ('' if pending else 'pp')
    assert db.query(m.PlPlanDeliveryProduct).count() == 1


def test_missing_link_must_not_guess_between_pending_and_linked_rows(db):
    pp, old = seed(db, 'delivery')
    old.unit = '吨'
    db.add(m.PlPlanDeliveryProduct(id='pending', tenant_no='t', plan_id='p', delivery_id='stop',
                                  plan_product_id='', product_id='sku', product_no='SKU',
                                  product_name='产品', unit='吨'))
    db.flush()
    with pytest.raises(HTTPException) as exc:
        replace_plan_stops(db, 't', 'p', None,
                           [PlanDeliveryBody(id='stop', products=[product(planProductId=None)])],
                           'editor', 'dept', [pp], preserve_pending_draft=True)
    assert exc.value.status_code == 422
    assert not db.dirty and not db.new