shanxia2.0/backend/tests/test_crm_child_differential_save.py

"""CRM edits retain child identities; all persistence uses an isolated SQLite database."""
from datetime import date, datetime
from decimal import Decimal

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

REAL_BRAND_RESOLVER = qp._resolve_line_brand_fields


@pytest.fixture
def db(monkeypatch):
    engine = create_engine('sqlite://')
    for model in (CrmContractProduct, CrmContractProductPriceTier, CrmQuoteProduct,
                  PlPlan, PlPlanLoadProduct, PlPlanDeliveryProduct):
        model.__table__.create(engine)
    monkeypatch.setattr(cp, 'get_product', lambda *a: None)
    monkeypatch.setattr(qp, 'get_product', lambda *a: None)
    monkeypatch.setattr(cp, '_resolve_line_brand_fields', lambda *a, **k: (None, None, None))
    monkeypatch.setattr(qp, '_resolve_line_brand_fields', lambda *a, **k: (None, None, None))
    with Session(engine) as session:
        yield session


def line(db, kind='contract', id='a', product='p', **extra):
    model = CrmContractProduct if kind == 'contract' else CrmQuoteProduct
    data = dict(id=id, tenant_no='t', product_id=product, product_name='Product',
                quantity=2, price=5, subtotal=10, unit='吨', package_spec='桶',
                create_by='creator', create_time=datetime(2020, 1, 1), del_flag=0)
    data[kind + '_id'] = 'parent'
    data.update(extra)
    obj = model(**data)
    db.add(obj); db.flush()
    return obj


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


@pytest.mark.parametrize('kind', ['contract', 'quote'])
def test_update_add_remove_retains_existing_identity(db, kind):
    a = line(db, kind); b = line(db, kind, id='b', product='p2')
    reconcile(db, kind, [{'id':'a', 'quantity':3}, {'productId':'new', 'productName':'New', 'quantity':1}])
    assert a.id == 'a' and a.del_flag == 0 and a.quantity == 3
    assert a.create_by == 'creator' and a.create_time == datetime(2020, 1, 1)
    assert b.del_flag == 1
    model = type(a)
    assert db.query(model).filter(model.del_flag == 0).count() == 2


@pytest.mark.parametrize('kind', ['contract', 'quote'])
@pytest.mark.parametrize('bad', [[{'id':'a','quantity':99},{'id':'foreign'}], [{'id':'a'},{'id':'a'}]])
def test_invalid_ids_reject_before_mutating(db, kind, bad):
    a = line(db, kind)
    line(db, kind, id='foreign', tenant_no='other')
    with pytest.raises(ValueError): reconcile(db, kind, bad)
    assert a.quantity == 2 and a.del_flag == 0 and not db.dirty


@pytest.mark.parametrize('kind', ['contract', 'quote'])
def test_legacy_unique_identity_and_ambiguous_duplicate(db, kind):
    a = line(db, kind)
    payload = {'productId':'p','unit':'吨','packageSpec':'桶','quantity':3}
    reconcile(db, kind, [payload])
    assert a.quantity == 3 and a.del_flag == 0
    b = line(db, kind, id='b')
    with pytest.raises(ValueError): reconcile(db, kind, [payload])
    assert a.del_flag == b.del_flag == 0


@pytest.mark.parametrize('kind', ['contract', 'quote'])
def test_explicit_existing_line_allows_adding_same_product_as_new_line(db, kind):
    a = line(db, kind)
    reconcile(db, kind, [{'id':'a'}, {'productId':'p','productName':'Product',
        'unit':'吨','packageSpec':'桶','quantity':5}])
    assert a.del_flag == 0 and a.quantity == 2
    rows = db.query(type(a)).filter(type(a).del_flag == 0).all()
    assert len(rows) == 2 and len({r.id for r in rows}) == 2


def test_contract_preserves_omitted_delivery_and_tiers(db):
    a = line(db, delivered_quantity=1, delivery_status='partial')
    tier = CrmContractProductPriceTier(id='tier',contract_id='parent',contract_product_id='a',
        tenant_no='t',start_date=date(2026,1,1),end_date=date(2026,2,1),price=3)
    db.add(tier); db.flush()
    reconcile(db, 'contract', [{'id':'a','quantity':4}])
    assert a.delivered_quantity == 1 and a.delivery_status == 'partial'
    assert tier.del_flag == 0
    reconcile(db, 'contract', [{'id':'a','priceTiers':[{'id':'tier','price':7}]}])
    assert tier.price == 7 and tier.id == 'tier' and tier.del_flag == 0
    reconcile(db, 'contract', [{'id':'a','priceTiers':[]}])
    assert tier.del_flag == 1


def test_foreign_tier_rejects_before_any_product_mutation(db):
    a = line(db)
    with pytest.raises(ValueError):
        reconcile(db, 'contract', [{'id':'a','quantity':9,'priceTiers':[{'id':'foreign'}]}])
    assert a.quantity == 2 and not db.dirty


@pytest.mark.parametrize('kind', ['contract', 'quote'])
def test_explicit_null_clears_nullable_text_and_brand(db, monkeypatch, kind):
    a = line(db, kind, description='old', brand_mode='required', brand_id='brand', brand_name='Brand')
    monkeypatch.setattr(qp, 'product_has_brands', lambda *a: True)
    # Restore the real resolver while keeping unrelated master-product access isolated.
    resolver = REAL_BRAND_RESOLVER
    monkeypatch.setattr(qp, 'product_has_brands', lambda *a: True)
    monkeypatch.setattr(qp, '_resolve_line_brand_fields', resolver)
    monkeypatch.setattr(cp, '_resolve_line_brand_fields', resolver)
    reconcile(db, kind, [{'id':'a','description':None,'brandMode':None,'brandId':None,'brandName':None}])
    assert a.description is None and a.brand_id is None and a.brand_mode == 'any'


def test_null_date_tier_remark_and_new_tier_default_order(db):
    a = line(db, delivery_date=date(2026,1,1))
    tier = CrmContractProductPriceTier(id='tier',contract_id='parent',contract_product_id='a',
        tenant_no='t',start_date=date(2026,1,1),end_date=date(2026,2,1),price=3,remark='old')
    db.add(tier); db.flush()
    reconcile(db, 'contract', [{'id':'a','deliveryDate':None,'priceTiers':[
        {'id':'tier','remark':None},
        {'startDate':'2026-03-01','endDate':'2026-04-01','price':4},
        {'startDate':'2026-05-01','endDate':'2026-06-01','price':5}]}])
    assert a.delivery_date is None and tier.remark is None
    added = db.query(CrmContractProductPriceTier).filter(CrmContractProductPriceTier.id != 'tier').order_by(CrmContractProductPriceTier.start_date).all()
    assert [t.sort for t in added] == [2,3]


@pytest.mark.parametrize('archive', ['draft', 'archived'])
def test_contract_referenced_by_valid_plan_cannot_be_removed(db, archive):
    a = line(db)
    db.add(PlPlan(id='plan',plan_no='P',create_date=date.today(),plan_date=date.today(),
                  tenant_no='t',belongs_company='company',plan_type='sales',follow_status='arranged',archive_status=archive))
    db.add(PlPlanDeliveryProduct(id='delivery-product',delivery_id='delivery',plan_id='plan',
        plan_product_id='plan-product',product_id='p',product_no='P',product_name='Product',
        contract_id='parent',contract_product_id='a',tenant_no='t'))
    db.flush()
    with pytest.raises(ValueError, match='引用'): reconcile(db, 'contract', [])
    assert a.del_flag == 0


@pytest.mark.parametrize('kind', ['contract', 'quote'])
def test_none_means_no_child_write(db, kind):
    a = line(db, kind)
    reconcile(db, kind, None)
    assert a.del_flag == 0 and not db.dirty


@pytest.mark.parametrize('schema', [ContractUpdateBody, QuoteUpdateBody])
def test_update_schema_distinguishes_omitted_products(schema):
    body = schema(id='parent', customerId='customer', contractType='sale')
    assert body.products is None
    assert 'products' not in body.model_dump(by_alias=True, exclude_unset=True)


@pytest.mark.parametrize('kind', ['contract', 'quote'])
def test_whole_document_update_uses_diff_and_validates_before_parent_changes(db, monkeypatch, kind):
    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: '')
    a = line(db, kind)
    update = getattr(module, 'update_' + kind)
    original_status = parent.status
    with pytest.raises(ValueError):
        update(db, 'parent', tenant_no='t', data={'products':[{'id':'foreign'}]})
    assert parent.status == original_status
    update(db, 'parent', tenant_no='t', data={'products':[{'id':'a','quantity':4}]})
    assert a.del_flag == 0 and a.id == 'a' and a.quantity == 4
    assert a.subtotal == 20