265 lines
12 KiB
Python
265 lines
12 KiB
Python
"""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
|