shanxia2.0/backend/tests/test_crm_missing_identity_patch.py

"""CRM missing-ID partial saves: real SQLite child writes, forbidden business MySQL."""
from datetime import date, datetime

import pymysql
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from app.crud.crm import contract_product as cp, quote_product as qp
from app.models.crm_models import CrmContractProduct, CrmContractProductPriceTier, CrmQuoteProduct
from app.models.pl_models import PlPlan, PlPlanLoadProduct, PlPlanDeliveryProduct
from app.schemas.crm.contract import ContractUpdateBody
from app.schemas.crm.quote import QuoteUpdateBody


@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 model in (CrmContractProduct, CrmContractProductPriceTier, CrmQuoteProduct,
                  PlPlan, PlPlanLoadProduct, PlPlanDeliveryProduct):
        model.__table__.create(engine)
    # Product-library enrichment is outside identity reconciliation.
    monkeypatch.setattr(cp, 'get_product', lambda *args: None)
    monkeypatch.setattr(qp, 'get_product', lambda *args: None)
    def brand_fields(*args, **kwargs):
        data = kwargs['data']
        keys = ('brandMode', 'brandId', 'brandName', 'brand_mode', 'brand_id', 'brand_name')
        if not any(key in data for key in keys):
            return kwargs.get('defaults', (None, None, None))
        return tuple(qp._get_field(data, snake, camel) for snake, camel in
                     (('brand_mode', 'brandMode'), ('brand_id', 'brandId'), ('brand_name', 'brandName')))
    for module in (cp, qp):
        monkeypatch.setattr(module, '_resolve_line_brand_fields', brand_fields)
    with Session(engine, autoflush=False) as session:
        yield session
    engine.dispose()


@pytest.fixture(params=['quote', 'contract'])
def kind(request):
    return request.param


def line(db, kind, row_id='old', **extra):
    cls = CrmContractProduct if kind == 'contract' else CrmQuoteProduct
    values = dict(id=row_id, tenant_no='t', product_id='sku', product_name='Product',
                  model='A', unit='吨', package_spec='桶', quantity=2, price=5, subtotal=10,
                  brand_mode='required', brand_id='brand', brand_name='Brand', create_by='creator',
                  create_time=datetime(2020, 1, 2), del_flag=0)
    values[kind + '_id'] = 'parent'
    values.update(extra)
    row = cls(**values)
    db.add(row)
    db.flush()
    return row


def product(**kwargs):
    return dict(productId='sku', productName='Product', **kwargs)


def sync(db, kind, products):
    module = cp if kind == 'contract' else qp
    getattr(module, 'sync_' + kind + '_products')(
        db, tenant_no='t', **{kind + '_id': 'parent'}, products=products, user_id='editor')
    db.flush()


def test_omitted_identity_fields_reuse_unique_old_line_twice(db, kind):
    old = line(db, kind)
    for quantity in (3, 4):
        sync(db, kind, [product(quantity=quantity)])
    assert db.query(type(old)).count() == 1
    assert old.del_flag == 0 and old.quantity == 4 and old.subtotal == 20
    assert (old.model, old.unit, old.package_spec, old.brand_id) == ('A', '吨', '桶', 'brand')
    assert old.create_by == 'creator' and old.create_time == datetime(2020, 1, 2)


@pytest.mark.parametrize('field', ['unit', 'model', 'brandId', 'brandMode', 'packageSpec'])
def test_each_omitted_key_is_not_defaulted_for_matching(db, kind, field):
    old = line(db, kind)
    body = product(unit='吨', model='A', brandId='brand', brandMode='required', packageSpec='桶', quantity=4)
    body.pop(field)
    sync(db, kind, [body])
    assert old.del_flag == 0 and old.quantity == 4
    assert db.query(type(old)).count() == 1


@pytest.mark.parametrize('discriminator,old_fields,first_value,second_value', [
    ('unit', 'unit', '吨', '千克'), ('model', 'model', 'A', 'B'),
    ('brandId', 'brand_id', 'brand', 'brand2'), ('packageSpec', 'package_spec', '桶', '袋'),
])
def test_same_sku_variants_match_submitted_fields_after_reordering(db, kind, discriminator, old_fields, first_value, second_value):
    first = line(db, kind, **{old_fields: first_value})
    second = line(db, kind, 'second', **{old_fields: second_value})
    sync(db, kind, [product(**{discriminator: second_value}, quantity=6),
                    product(**{discriminator: first_value}, quantity=7)])
    assert first.quantity == 7 and second.quantity == 6
    assert first.del_flag == second.del_flag == 0 and db.query(type(first)).count() == 2


@pytest.mark.parametrize('payload', [[product()], [product(), product()]])
def test_ambiguous_or_competing_anonymous_rows_reject_before_writes(db, kind, payload):
    old = line(db, kind)
    if len(payload) == 1:
        line(db, kind, 'second', model='B')
    with pytest.raises(ValueError):
        sync(db, kind, payload)
    assert not db.new and not db.dirty and old.del_flag == 0


@pytest.mark.parametrize('product_id', [None, '', '  '])
def test_missing_product_id_cannot_guess_from_name(db, kind, product_id):
    old = line(db, kind)
    with pytest.raises(ValueError):
        sync(db, kind, [{'productName': 'Product', 'productId': product_id}])
    assert not db.new and not db.dirty and old.del_flag == 0


def test_explicit_id_reserved_before_same_product_anonymous_new_row(db, kind):
    old = line(db, kind)
    sync(db, kind, [product(quantity=3), {'id': 'old', 'quantity': 4}])
    assert old.del_flag == 0 and old.quantity == 4
    assert db.query(type(old)).count() == 2


@pytest.mark.parametrize('field', ['unit', 'model', 'brandId', 'packageSpec'])
def test_explicit_null_is_a_submitted_key_not_an_omission(db, kind, field):
    old = line(db, kind)
    mapping = {'unit': 'unit', 'model': 'model', 'brandId': 'brand_id', 'packageSpec': 'package_spec'}
    empty = line(db, kind, 'empty', **{mapping[field]: None})
    sync(db, kind, [{'id': 'old'}, product(**{field: None}, quantity=8)])
    assert old.del_flag == empty.del_flag == 0 and empty.quantity == 8
    assert db.query(type(old)).count() == 2


def test_explicit_id_can_clear_nullable_identity_fields_without_new_line(db, kind):
    old = line(db, kind)
    sync(db, kind, [{'id': 'old', 'model': None, 'brandMode': None, 'brandId': None, 'brandName': None}])
    assert old.del_flag == 0 and not old.model and old.brand_id is None
    assert db.query(type(old)).count() == 1


def test_omitted_collection_preserves_and_empty_deletes(db, kind):
    old = line(db, kind)
    sync(db, kind, None)
    assert old.del_flag == 0
    sync(db, kind, [])
    assert old.del_flag == 1


def test_new_different_product_and_remove_old(db, kind):
    old = line(db, kind)
    sync(db, kind, [{'productId': 'other', 'productName': 'Other', 'quantity': 3}])
    assert old.del_flag == 1 and db.query(type(old)).filter_by(del_flag=0).one().product_id == 'other'


@pytest.mark.parametrize('scope', [{'tenant_no': 'other'}, {'del_flag': 1}, {'parent': 'other'}])
def test_explicit_id_must_be_active_and_owned(db, kind, scope):
    values = dict(scope)
    if 'parent' in values:
        values[kind + '_id'] = values.pop('parent')
    line(db, kind, **values)
    with pytest.raises(ValueError):
        sync(db, kind, [{'id': 'old', 'quantity': 4}])
    assert not db.new and not db.dirty


def test_duplicate_explicit_id_rejected_before_writes(db, kind):
    old = line(db, kind)
    with pytest.raises(ValueError):
        sync(db, kind, [{'id': 'old', 'quantity': 9}, {'id': 'old'}])
    assert not db.new and not db.dirty and old.quantity == 2


def test_contract_downstream_reference_allows_partial_match_but_not_removal(db):
    old = line(db, 'contract', delivered_quantity=1, delivery_status='partial')
    db.add(PlPlan(id='plan', tenant_no='t', plan_no='P', follow_status='completed', archive_status='archived', create_date=date(2026, 1, 1), belongs_company='shanxia', plan_type='outbound'))
    db.add(PlPlanDeliveryProduct(id='ref', tenant_no='t', plan_id='plan', delivery_id='stop',
                                contract_product_id='old', plan_product_id='pp', product_id='sku', product_no='SKU', product_name='Product'))
    db.flush()
    sync(db, 'contract', [product(quantity=4)])
    assert old.del_flag == 0 and old.delivered_quantity == 1 and old.delivery_status == 'partial'
    db.commit()
    with pytest.raises(ValueError):
        sync(db, 'contract', [])
    assert old.del_flag == 0 and not db.new and not db.dirty


def test_tier_natural_key_matching_and_parent_partial_identity_are_independent(db):
    old = line(db, 'contract')
    tier = CrmContractProductPriceTier(id='tier', tenant_no='t', contract_id='parent',
        contract_product_id='old', start_date=date(2026, 1, 1), end_date=date(2026, 2, 1), price=2)
    db.add(tier)
    db.flush()
    sync(db, 'contract', [product(priceTiers=[{'startDate': '2026-01-01', 'endDate': '2026-02-01', 'price': 3}])])
    assert old.del_flag == tier.del_flag == 0 and tier.price == 3
    assert db.query(CrmContractProductPriceTier).count() == 1
    db.commit()
    with pytest.raises(ValueError):
        sync(db, 'contract', [product(priceTiers=[{'startDate': '2026-01-01', 'price': 4}])])
    assert not db.new and not db.dirty


@pytest.fixture
def save_whole(db, kind, monkeypatch):
    from app.crud.crm import contract, quote
    from app.models.crm_models import CrmContract, CrmQuote
    module = contract if kind == 'contract' else quote
    parent = (CrmContract(id='parent', contract_type='sale', status='executing', customer_id='c',
                          contract_no='', contract_name='Contract', amount=10) if kind == 'contract'
              else CrmQuote(id='parent', quote_type='sales', status='accepted', customer_id='c'))
    getter = '_get_contract_row_in_scope' if kind == 'contract' else '_get_quote_in_scope'
    monkeypatch.setattr(module, getter, lambda *a, **k: parent)
    for name in ('invalidate_all_approvals', 'assert_products_allowed_for_supplier',
                 '_resolve_' + kind + '_product_ids', 'sync_approval_notifications'):
        monkeypatch.setattr(module, name, lambda *a, **k: None)
    monkeypatch.setattr(module, 'get_' + kind + '_detail', lambda *a, **k: {})
    if kind == 'contract':
        monkeypatch.setattr(module, '_apply_contract_entity_fields', lambda *a, **k: None)
        monkeypatch.setattr(module, '_build_contract_name_from_parties', lambda *a, **k: 'Contract')
    else:
        for name in ('_enrich_contact_fields', '_enrich_supplier_name', 'calculate_quote_amount'):
            monkeypatch.setattr(module, name, lambda *a, **k: None)
        monkeypatch.setattr(module, 'get_customer', lambda *a, **k: None)
        monkeypatch.setattr(module, '_get_supplier_short_name', lambda *a, **k: '')
    def save(products):
        schema = ContractUpdateBody if kind == 'contract' else QuoteUpdateBody
        data = schema(id='parent', customerId='c', contractType='sale', products=products).model_dump(by_alias=True, exclude_unset=True)
        data.pop('id')
        return getattr(module, 'update_' + kind)(db, 'parent', tenant_no='t', data=data)
    return save, parent


def test_schema_to_whole_update_keeps_missing_unit_and_other_fields(db, kind, save_whole):
    old = line(db, kind)
    save, parent = save_whole
    for _ in range(2):
        save([product(model='A', packageSpec='桶', quantity=3, price=5)])
    assert old.del_flag == 0 and old.unit == '吨' and old.brand_id == 'brand'
    assert db.query(type(old)).count() == 1 and old.quantity == 3


def test_whole_ambiguous_partial_identity_rejected_before_parent_status_changes(db, kind, save_whole):
    line(db, kind)
    line(db, kind, 'second', model='B')
    save, parent = save_whole
    status = parent.status
    with pytest.raises(ValueError):
        save([product(packageSpec='桶', quantity=3)])
    assert parent.status == status and not db.new and not db.dirty


@pytest.mark.parametrize('field', ['unit', 'model', 'brandId', 'packageSpec'])
def test_explicit_null_disambiguates_empty_key_from_nonempty_key(db, kind, field):
    old = line(db, kind)
    mapping = {'unit': 'unit', 'model': 'model', 'brandId': 'brand_id', 'packageSpec': 'package_spec'}
    empty = line(db, kind, 'empty', **{mapping[field]: None})
    sync(db, kind, [product(**{field: None}, quantity=8)])
    assert old.del_flag == 1 and empty.del_flag == 0 and empty.quantity == 8
    assert db.query(type(old)).count() == 2