test(business-rules): add 44 regression tests for kit pricing, discount, and SKU mapping
38 unit tests (no Oracle) covering: discount VAT split, build_articles_json, kit detection pattern, sync_prices skip logic, VAT included normalization, validate_kit_component_prices (pret=0 allowed), dual policy assignment, and resolve_codmat_ids deduplication. 6 Oracle integration tests covering: multi-kit discount merge, per-kit discount placement, distributed mode total, markup no negative discount, price=0 component import, and duplicate CODMAT different prices. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Agent
494
api/tests/test_business_rules.py
Normal file
494
api/tests/test_business_rules.py
Normal file
@@ -0,0 +1,494 @@
|
||||
"""
|
||||
Business Rule Regression Tests
|
||||
==============================
|
||||
Regression tests for historical bug fixes in kit pricing, discount calculation,
|
||||
duplicate CODMAT resolution, price sync, and VAT normalization.
|
||||
|
||||
Run:
|
||||
cd api && python -m pytest tests/test_business_rules.py -v
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
# --- Set env vars BEFORE any app import ---
|
||||
_tmpdir = tempfile.mkdtemp()
|
||||
_sqlite_path = os.path.join(_tmpdir, "test_biz.db")
|
||||
|
||||
os.environ["FORCE_THIN_MODE"] = "true"
|
||||
os.environ["SQLITE_DB_PATH"] = _sqlite_path
|
||||
os.environ["ORACLE_DSN"] = "dummy"
|
||||
os.environ["ORACLE_USER"] = "dummy"
|
||||
os.environ["ORACLE_PASSWORD"] = "dummy"
|
||||
os.environ["JSON_OUTPUT_DIR"] = _tmpdir
|
||||
|
||||
_api_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
if _api_dir not in sys.path:
|
||||
sys.path.insert(0, _api_dir)
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from app.services.import_service import build_articles_json, compute_discount_split
|
||||
from app.services.order_reader import OrderData, OrderItem
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def make_item(sku="SKU1", price=100.0, quantity=1, vat=19):
|
||||
return OrderItem(sku=sku, name=f"Product {sku}", price=price, quantity=quantity, vat=vat)
|
||||
|
||||
|
||||
def make_order(items, discount_total=0.0, delivery_cost=0.0, discount_vat=None):
|
||||
order = OrderData(
|
||||
id="1", number="TEST-001", date="2026-01-01",
|
||||
items=items, discount_total=discount_total,
|
||||
delivery_cost=delivery_cost,
|
||||
)
|
||||
if discount_vat is not None:
|
||||
order.discount_vat = discount_vat
|
||||
return order
|
||||
|
||||
|
||||
def is_kit(comps):
|
||||
"""Kit detection pattern used in validation_service and price_sync_service."""
|
||||
return len(comps) > 1 or (
|
||||
len(comps) == 1 and (comps[0].get("cantitate_roa") or 1) > 1
|
||||
)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group 1: compute_discount_split()
|
||||
# ===========================================================================
|
||||
|
||||
class TestDiscountSplit:
|
||||
"""Regression: discount split by VAT rate (import_service.py:63)."""
|
||||
|
||||
def test_single_vat_rate(self):
|
||||
order = make_order([make_item(vat=19), make_item("SKU2", vat=19)], discount_total=10.0)
|
||||
result = compute_discount_split(order, {"split_discount_vat": "1"})
|
||||
assert result == {"19": 10.0}
|
||||
|
||||
def test_multiple_vat_proportional(self):
|
||||
items = [make_item("A", price=100, quantity=1, vat=19),
|
||||
make_item("B", price=50, quantity=1, vat=9)]
|
||||
order = make_order(items, discount_total=15.0)
|
||||
result = compute_discount_split(order, {"split_discount_vat": "1"})
|
||||
assert result == {"9": 5.0, "19": 10.0}
|
||||
|
||||
def test_zero_returns_none(self):
|
||||
order = make_order([make_item()], discount_total=0)
|
||||
assert compute_discount_split(order, {"split_discount_vat": "1"}) is None
|
||||
|
||||
def test_zero_price_items_excluded(self):
|
||||
items = [make_item("A", price=0, quantity=1, vat=19),
|
||||
make_item("B", price=100, quantity=2, vat=9)]
|
||||
order = make_order(items, discount_total=5.0)
|
||||
result = compute_discount_split(order, {"split_discount_vat": "1"})
|
||||
assert result == {"9": 5.0}
|
||||
|
||||
def test_disabled_multiple_rates(self):
|
||||
items = [make_item("A", vat=19), make_item("B", vat=9)]
|
||||
order = make_order(items, discount_total=10.0)
|
||||
result = compute_discount_split(order, {"split_discount_vat": "0"})
|
||||
assert result is None
|
||||
|
||||
def test_rounding_remainder(self):
|
||||
items = [make_item("A", price=33.33, quantity=1, vat=19),
|
||||
make_item("B", price=33.33, quantity=1, vat=9),
|
||||
make_item("C", price=33.34, quantity=1, vat=5)]
|
||||
order = make_order(items, discount_total=10.0)
|
||||
result = compute_discount_split(order, {"split_discount_vat": "1"})
|
||||
assert result is not None
|
||||
assert abs(sum(result.values()) - 10.0) < 0.001
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group 2: build_articles_json()
|
||||
# ===========================================================================
|
||||
|
||||
class TestBuildArticlesJson:
|
||||
"""Regression: discount lines, policy bridge, transport (import_service.py:117)."""
|
||||
|
||||
def test_discount_line_negative_quantity(self):
|
||||
items = [make_item()]
|
||||
order = make_order(items, discount_total=5.0)
|
||||
settings = {"discount_codmat": "DISC01", "split_discount_vat": "0"}
|
||||
result = json.loads(build_articles_json(items, order, settings))
|
||||
disc_lines = [a for a in result if a["sku"] == "DISC01"]
|
||||
assert len(disc_lines) == 1
|
||||
assert disc_lines[0]["quantity"] == "-1"
|
||||
assert disc_lines[0]["price"] == "5.0"
|
||||
|
||||
def test_discount_uses_actual_vat_not_21(self):
|
||||
items = [make_item("A", vat=9), make_item("B", vat=9)]
|
||||
order = make_order(items, discount_total=3.0)
|
||||
settings = {"discount_codmat": "DISC01", "split_discount_vat": "1"}
|
||||
result = json.loads(build_articles_json(items, order, settings))
|
||||
disc_lines = [a for a in result if a["sku"] == "DISC01"]
|
||||
assert len(disc_lines) == 1
|
||||
assert disc_lines[0]["vat"] == "9"
|
||||
|
||||
def test_discount_multi_vat_creates_multiple_lines(self):
|
||||
items = [make_item("A", price=100, vat=19), make_item("B", price=50, vat=9)]
|
||||
order = make_order(items, discount_total=15.0)
|
||||
settings = {"discount_codmat": "DISC01", "split_discount_vat": "1"}
|
||||
result = json.loads(build_articles_json(items, order, settings))
|
||||
disc_lines = [a for a in result if a["sku"] == "DISC01"]
|
||||
assert len(disc_lines) == 2
|
||||
vats = {d["vat"] for d in disc_lines}
|
||||
assert "9" in vats
|
||||
assert "19" in vats
|
||||
|
||||
def test_discount_fallback_uses_gomag_vat(self):
|
||||
items = [make_item("A", vat=19), make_item("B", vat=9)]
|
||||
order = make_order(items, discount_total=5.0, discount_vat="9")
|
||||
settings = {"discount_codmat": "DISC01", "split_discount_vat": "0"}
|
||||
result = json.loads(build_articles_json(items, order, settings))
|
||||
disc_lines = [a for a in result if a["sku"] == "DISC01"]
|
||||
assert len(disc_lines) == 1
|
||||
assert disc_lines[0]["vat"] == "9"
|
||||
|
||||
def test_per_article_policy_bridge(self):
|
||||
items = [make_item("SKU1")]
|
||||
settings = {"_codmat_policy_map": {"SKU1": 42}, "id_pol": "1"}
|
||||
result = json.loads(build_articles_json(items, settings=settings))
|
||||
assert result[0]["id_pol"] == "42"
|
||||
|
||||
def test_policy_same_as_default_omitted(self):
|
||||
items = [make_item("SKU1")]
|
||||
settings = {"_codmat_policy_map": {"SKU1": 1}, "id_pol": "1"}
|
||||
result = json.loads(build_articles_json(items, settings=settings))
|
||||
assert "id_pol" not in result[0]
|
||||
|
||||
def test_transport_line_added(self):
|
||||
items = [make_item()]
|
||||
order = make_order(items, delivery_cost=15.0)
|
||||
settings = {"transport_codmat": "TR", "transport_vat": "19"}
|
||||
result = json.loads(build_articles_json(items, order, settings))
|
||||
tr_lines = [a for a in result if a["sku"] == "TR"]
|
||||
assert len(tr_lines) == 1
|
||||
assert tr_lines[0]["quantity"] == "1"
|
||||
assert tr_lines[0]["price"] == "15.0"
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group 3: Kit Detection Pattern
|
||||
# ===========================================================================
|
||||
|
||||
class TestKitDetection:
|
||||
"""Regression: kit detection for single-component repackaging (multiple code locations)."""
|
||||
|
||||
def test_multi_component(self):
|
||||
comps = [{"codmat": "A", "cantitate_roa": 1}, {"codmat": "B", "cantitate_roa": 1}]
|
||||
assert is_kit(comps) is True
|
||||
|
||||
def test_single_component_repackaging(self):
|
||||
comps = [{"codmat": "CAF01", "cantitate_roa": 10}]
|
||||
assert is_kit(comps) is True
|
||||
|
||||
def test_true_1to1_not_kit(self):
|
||||
comps = [{"codmat": "X", "cantitate_roa": 1}]
|
||||
assert is_kit(comps) is False
|
||||
|
||||
def test_none_cantitate_treated_as_1(self):
|
||||
comps = [{"codmat": "X", "cantitate_roa": None}]
|
||||
assert is_kit(comps) is False
|
||||
|
||||
def test_empty_components(self):
|
||||
assert is_kit([]) is False
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group 4: sync_prices_from_order() — Kit Skip Logic
|
||||
# ===========================================================================
|
||||
|
||||
class TestSyncPricesKitSkip:
|
||||
"""Regression: kit SKUs must be skipped in order-based price sync."""
|
||||
|
||||
def _make_mock_order(self, sku, price=50.0):
|
||||
mock_order = MagicMock()
|
||||
mock_item = MagicMock()
|
||||
mock_item.sku = sku
|
||||
mock_item.price = price
|
||||
mock_order.items = [mock_item]
|
||||
return mock_order
|
||||
|
||||
@patch("app.services.validation_service.compare_and_update_price")
|
||||
def test_skips_multi_component_kit(self, mock_compare):
|
||||
from app.services.validation_service import sync_prices_from_order
|
||||
orders = [self._make_mock_order("KIT01")]
|
||||
mapped = {"KIT01": [
|
||||
{"codmat": "A", "id_articol": 1, "cantitate_roa": 1},
|
||||
{"codmat": "B", "id_articol": 2, "cantitate_roa": 1},
|
||||
]}
|
||||
mock_conn = MagicMock()
|
||||
sync_prices_from_order(orders, mapped, {}, {}, 1, conn=mock_conn,
|
||||
settings={"price_sync_enabled": "1"})
|
||||
mock_compare.assert_not_called()
|
||||
|
||||
@patch("app.services.validation_service.compare_and_update_price")
|
||||
def test_skips_repackaging_kit(self, mock_compare):
|
||||
from app.services.validation_service import sync_prices_from_order
|
||||
orders = [self._make_mock_order("CAFE100")]
|
||||
mapped = {"CAFE100": [
|
||||
{"codmat": "CAF01", "id_articol": 1, "cantitate_roa": 10},
|
||||
]}
|
||||
mock_conn = MagicMock()
|
||||
sync_prices_from_order(orders, mapped, {}, {}, 1, conn=mock_conn,
|
||||
settings={"price_sync_enabled": "1"})
|
||||
mock_compare.assert_not_called()
|
||||
|
||||
@patch("app.services.validation_service.compare_and_update_price")
|
||||
def test_processes_1to1_mapping(self, mock_compare):
|
||||
from app.services.validation_service import sync_prices_from_order
|
||||
mock_compare.return_value = {"updated": False, "old_price": 50.0, "new_price": 50.0, "codmat": "X"}
|
||||
orders = [self._make_mock_order("SKU1", price=50.0)]
|
||||
mapped = {"SKU1": [
|
||||
{"codmat": "X", "id_articol": 100, "cantitate_roa": 1, "cont": "371"},
|
||||
]}
|
||||
mock_conn = MagicMock()
|
||||
sync_prices_from_order(orders, mapped, {}, {"SKU1": 1}, 1, conn=mock_conn,
|
||||
settings={"price_sync_enabled": "1"})
|
||||
mock_compare.assert_called_once()
|
||||
call_args = mock_compare.call_args
|
||||
assert call_args[0][0] == 100 # id_articol
|
||||
assert call_args[0][2] == 50.0 # price
|
||||
|
||||
@patch("app.services.validation_service.compare_and_update_price")
|
||||
def test_skips_transport_discount_codmats(self, mock_compare):
|
||||
from app.services.validation_service import sync_prices_from_order
|
||||
orders = [self._make_mock_order("TRANSP", price=15.0)]
|
||||
mock_conn = MagicMock()
|
||||
sync_prices_from_order(orders, {}, {"TRANSP": {"id_articol": 99}}, {}, 1,
|
||||
conn=mock_conn,
|
||||
settings={"price_sync_enabled": "1",
|
||||
"transport_codmat": "TRANSP",
|
||||
"discount_codmat": "DISC"})
|
||||
mock_compare.assert_not_called()
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group 5: Kit Component with Own Mapping
|
||||
# ===========================================================================
|
||||
|
||||
class TestKitComponentOwnMapping:
|
||||
"""Regression: price_sync_service skips kit components that have their own ARTICOLE_TERTI mapping."""
|
||||
|
||||
def test_component_with_own_mapping_skipped(self):
|
||||
"""If comp_codmat is itself a key in mapped_data, it's skipped."""
|
||||
mapped_data = {
|
||||
"PACK-A": [{"codmat": "COMP-X", "id_articol": 1, "cantitate_roa": 1, "cont": "371"}],
|
||||
"COMP-X": [{"codmat": "COMP-X", "id_articol": 1, "cantitate_roa": 1, "cont": "371"}],
|
||||
}
|
||||
# The check is: if comp_codmat in mapped_data: continue
|
||||
comp_codmat = "COMP-X"
|
||||
assert comp_codmat in mapped_data # Should be skipped
|
||||
|
||||
def test_component_without_own_mapping_processed(self):
|
||||
"""If comp_codmat is NOT in mapped_data, it should be processed."""
|
||||
mapped_data = {
|
||||
"PACK-A": [{"codmat": "COMP-Y", "id_articol": 2, "cantitate_roa": 1, "cont": "371"}],
|
||||
}
|
||||
comp_codmat = "COMP-Y"
|
||||
assert comp_codmat not in mapped_data # Should be processed
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group 6: VAT Included Type Normalization
|
||||
# ===========================================================================
|
||||
|
||||
class TestVatIncludedNormalization:
|
||||
"""Regression: GoMag returns vat_included as int 1 or string '1' (price_sync_service.py:144)."""
|
||||
|
||||
def _compute_price_cu_tva(self, product):
|
||||
price = float(product.get("price", "0"))
|
||||
vat = float(product.get("vat", "19"))
|
||||
if str(product.get("vat_included", "1")) == "1":
|
||||
return price
|
||||
else:
|
||||
return price * (1 + vat / 100)
|
||||
|
||||
def test_vat_included_int_1(self):
|
||||
result = self._compute_price_cu_tva({"price": "100", "vat": "19", "vat_included": 1})
|
||||
assert result == 100.0
|
||||
|
||||
def test_vat_included_str_1(self):
|
||||
result = self._compute_price_cu_tva({"price": "100", "vat": "19", "vat_included": "1"})
|
||||
assert result == 100.0
|
||||
|
||||
def test_vat_included_int_0(self):
|
||||
result = self._compute_price_cu_tva({"price": "100", "vat": "19", "vat_included": 0})
|
||||
assert result == 119.0
|
||||
|
||||
def test_vat_included_str_0(self):
|
||||
result = self._compute_price_cu_tva({"price": "100", "vat": "19", "vat_included": "0"})
|
||||
assert result == 119.0
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group 7: validate_kit_component_prices — pret=0 allowed
|
||||
# ===========================================================================
|
||||
|
||||
class TestKitComponentPriceValidation:
|
||||
"""Regression: pret=0 in CRM is valid for kit components (validation_service.py:469)."""
|
||||
|
||||
def _call_validate(self, fetchone_returns):
|
||||
from app.services.validation_service import validate_kit_component_prices
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_cursor = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
mock_cursor.fetchone.return_value = fetchone_returns
|
||||
|
||||
mapped = {"KIT-SKU": [
|
||||
{"codmat": "COMP1", "id_articol": 100, "cont": "371", "cantitate_roa": 5},
|
||||
]}
|
||||
return validate_kit_component_prices(mapped, id_pol=1, conn=mock_conn)
|
||||
|
||||
def test_price_zero_not_rejected(self):
|
||||
result = self._call_validate((0,))
|
||||
assert result == {}
|
||||
|
||||
def test_missing_entry_rejected(self):
|
||||
result = self._call_validate(None)
|
||||
assert "KIT-SKU" in result
|
||||
assert "COMP1" in result["KIT-SKU"]
|
||||
|
||||
def test_skips_true_1to1(self):
|
||||
from app.services.validation_service import validate_kit_component_prices
|
||||
mock_conn = MagicMock()
|
||||
mapped = {"SKU1": [
|
||||
{"codmat": "X", "id_articol": 1, "cont": "371", "cantitate_roa": 1},
|
||||
]}
|
||||
result = validate_kit_component_prices(mapped, id_pol=1, conn=mock_conn)
|
||||
assert result == {}
|
||||
|
||||
def test_checks_repackaging(self):
|
||||
"""Single component with cantitate_roa > 1 should be checked."""
|
||||
from app.services.validation_service import validate_kit_component_prices
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_cursor = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
mock_cursor.fetchone.return_value = (51.50,)
|
||||
|
||||
mapped = {"CAFE100": [
|
||||
{"codmat": "CAF01", "id_articol": 100, "cont": "371", "cantitate_roa": 10},
|
||||
]}
|
||||
result = validate_kit_component_prices(mapped, id_pol=1, conn=mock_conn)
|
||||
assert result == {}
|
||||
mock_cursor.execute.assert_called_once()
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group 8: Dual Policy Assignment
|
||||
# ===========================================================================
|
||||
|
||||
class TestDualPolicyAssignment:
|
||||
"""Regression: cont 341/345 → production policy, others → sales (validation_service.py:282)."""
|
||||
|
||||
def _call_dual(self, codmats, direct_id_map, cursor_rows):
|
||||
from app.services.validation_service import validate_and_ensure_prices_dual
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_cursor = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
# The code uses `for row in cur:` to iterate, not fetchall
|
||||
mock_cursor.__iter__ = MagicMock(return_value=iter(cursor_rows))
|
||||
# Mock ensure_prices to do nothing
|
||||
with patch("app.services.validation_service.ensure_prices"):
|
||||
return validate_and_ensure_prices_dual(
|
||||
codmats, id_pol_vanzare=1, id_pol_productie=2,
|
||||
conn=mock_conn, direct_id_map=direct_id_map
|
||||
)
|
||||
|
||||
def test_cont_341_production(self):
|
||||
result = self._call_dual(
|
||||
{"COD1"},
|
||||
{"COD1": {"id_articol": 100, "cont": "341"}},
|
||||
[] # no existing prices
|
||||
)
|
||||
assert result["COD1"] == 2 # id_pol_productie
|
||||
|
||||
def test_cont_345_production(self):
|
||||
result = self._call_dual(
|
||||
{"COD1"},
|
||||
{"COD1": {"id_articol": 100, "cont": "345"}},
|
||||
[]
|
||||
)
|
||||
assert result["COD1"] == 2
|
||||
|
||||
def test_other_cont_sales(self):
|
||||
result = self._call_dual(
|
||||
{"COD1"},
|
||||
{"COD1": {"id_articol": 100, "cont": "371"}},
|
||||
[]
|
||||
)
|
||||
assert result["COD1"] == 1 # id_pol_vanzare
|
||||
|
||||
def test_existing_sales_preferred(self):
|
||||
result = self._call_dual(
|
||||
{"COD1"},
|
||||
{"COD1": {"id_articol": 100, "cont": "345"}},
|
||||
[(100, 1), (100, 2)] # price exists in BOTH policies
|
||||
)
|
||||
assert result["COD1"] == 1 # sales preferred when both exist
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group 9: Duplicate CODMAT — resolve_codmat_ids
|
||||
# ===========================================================================
|
||||
|
||||
class TestResolveCodmatIds:
|
||||
"""Regression: ROW_NUMBER dedup returns exactly 1 id_articol per CODMAT."""
|
||||
|
||||
@patch("app.services.validation_service.database")
|
||||
def test_returns_one_per_codmat(self, mock_db):
|
||||
from app.services.validation_service import resolve_codmat_ids
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_cursor = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
# Simulate ROW_NUMBER already deduped: 1 row per codmat
|
||||
mock_cursor.__iter__ = MagicMock(return_value=iter([
|
||||
("COD1", 100, "345"),
|
||||
("COD2", 200, "341"),
|
||||
]))
|
||||
|
||||
result = resolve_codmat_ids({"COD1", "COD2"}, conn=mock_conn)
|
||||
assert len(result) == 2
|
||||
assert result["COD1"]["id_articol"] == 100
|
||||
assert result["COD2"]["id_articol"] == 200
|
||||
|
||||
@patch("app.services.validation_service.database")
|
||||
def test_resolve_mapped_one_per_sku_codmat(self, mock_db):
|
||||
from app.services.validation_service import resolve_mapped_codmats
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_cursor = MagicMock()
|
||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
||||
# 1 row per (sku, codmat) pair
|
||||
mock_cursor.__iter__ = MagicMock(return_value=iter([
|
||||
("SKU1", "COD1", 100, "345", 10),
|
||||
("SKU1", "COD2", 200, "341", 1),
|
||||
]))
|
||||
|
||||
result = resolve_mapped_codmats({"SKU1"}, mock_conn)
|
||||
assert "SKU1" in result
|
||||
assert len(result["SKU1"]) == 2
|
||||
codmats = [c["codmat"] for c in result["SKU1"]]
|
||||
assert "COD1" in codmats
|
||||
assert "COD2" in codmats
|
||||
@@ -528,6 +528,379 @@ def test_repackaging_kit_pricing():
|
||||
return False
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Group 10: Business Rule Regression Tests (Oracle integration)
|
||||
# ===========================================================================
|
||||
|
||||
def _create_test_partner(cur, suffix):
|
||||
"""Helper: create a test partner and return its ID."""
|
||||
partner_var = cur.var(oracledb.NUMBER)
|
||||
name = f'Test BizRule {suffix}'
|
||||
cur.execute("""
|
||||
DECLARE v_id NUMBER;
|
||||
BEGIN
|
||||
v_id := PACK_IMPORT_PARTENERI.cauta_sau_creeaza_partener(
|
||||
NULL, :name, 'JUD:Bucuresti;BUCURESTI;Str Test;1',
|
||||
'0720000000', 'bizrule@test.com');
|
||||
:result := v_id;
|
||||
END;
|
||||
""", {'name': name, 'result': partner_var})
|
||||
return partner_var.getvalue()
|
||||
|
||||
|
||||
def _import_order(cur, order_number, partner_id, articles_json, kit_mode='separate_line', id_pol=1):
|
||||
"""Helper: call importa_comanda and return order ID."""
|
||||
result_var = cur.var(oracledb.NUMBER)
|
||||
cur.execute("""
|
||||
DECLARE v_id NUMBER;
|
||||
BEGIN
|
||||
PACK_IMPORT_COMENZI.importa_comanda(
|
||||
:order_number, SYSDATE, :partner_id,
|
||||
:articles_json,
|
||||
NULL, NULL,
|
||||
:id_pol, NULL, NULL,
|
||||
:kit_mode,
|
||||
NULL, NULL, NULL,
|
||||
v_id);
|
||||
:result := v_id;
|
||||
END;
|
||||
""", {
|
||||
'order_number': order_number,
|
||||
'partner_id': partner_id,
|
||||
'articles_json': articles_json,
|
||||
'id_pol': id_pol,
|
||||
'kit_mode': kit_mode,
|
||||
'result': result_var
|
||||
})
|
||||
return result_var.getvalue()
|
||||
|
||||
|
||||
def _get_order_lines(cur, order_id):
|
||||
"""Helper: fetch COMENZI_ELEMENTE rows for an order."""
|
||||
cur.execute("""
|
||||
SELECT ce.CANTITATE, ce.PRET, na.CODMAT, ce.PTVA
|
||||
FROM COMENZI_ELEMENTE ce
|
||||
JOIN NOM_ARTICOLE na ON ce.ID_ARTICOL = na.ID_ARTICOL
|
||||
WHERE ce.ID_COMANDA = :oid
|
||||
ORDER BY ce.CANTITATE DESC, ce.PRET DESC
|
||||
""", {'oid': order_id})
|
||||
return cur.fetchall()
|
||||
|
||||
|
||||
def test_multi_kit_discount_merge():
|
||||
"""Regression (0666d6b): 2 identical kits at same VAT must merge discount lines,
|
||||
not crash on duplicate check collision."""
|
||||
print("\n" + "=" * 60)
|
||||
print("TEST: Multi-kit discount merge (separate_line)")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
suffix = f'{datetime.now().strftime("%H%M%S")}-{random.randint(1000, 9999)}'
|
||||
setup_test_data(cur)
|
||||
partner_id = _create_test_partner(cur, suffix)
|
||||
|
||||
# 2 identical CAFE100 kits: total web = 2 * 450 = 900
|
||||
articles_json = '[{"sku": "CAFE100", "cantitate": 2, "pret": 450}]'
|
||||
order_id = _import_order(cur, f'TEST-BIZ-MERGE-{suffix}', partner_id, articles_json)
|
||||
|
||||
assert order_id and order_id > 0, "Order import failed"
|
||||
rows = _get_order_lines(cur, order_id)
|
||||
|
||||
art_lines = [r for r in rows if r[0] > 0]
|
||||
disc_lines = [r for r in rows if r[0] < 0]
|
||||
assert len(art_lines) >= 1, f"Expected article line(s), got {len(art_lines)}"
|
||||
assert len(disc_lines) >= 1, f"Expected discount line(s), got {len(disc_lines)}"
|
||||
|
||||
total = sum(r[0] * r[1] for r in rows)
|
||||
expected = 900.0
|
||||
print(f" Total: {total:.2f} (expected: {expected:.2f})")
|
||||
assert abs(total - expected) < 0.02, f"Total {total:.2f} != expected {expected:.2f}"
|
||||
print(" PASS")
|
||||
|
||||
conn.commit()
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" FAIL: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def test_kit_discount_per_kit_placement():
|
||||
"""Regression (580ca59): discount lines must appear after article lines (both present)."""
|
||||
print("\n" + "=" * 60)
|
||||
print("TEST: Kit discount per-kit placement")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
suffix = f'{datetime.now().strftime("%H%M%S")}-{random.randint(1000, 9999)}'
|
||||
setup_test_data(cur)
|
||||
partner_id = _create_test_partner(cur, suffix)
|
||||
|
||||
articles_json = '[{"sku": "CAFE100", "cantitate": 1, "pret": 450}]'
|
||||
order_id = _import_order(cur, f'TEST-BIZ-PLACE-{suffix}', partner_id, articles_json)
|
||||
|
||||
assert order_id and order_id > 0, "Order import failed"
|
||||
rows = _get_order_lines(cur, order_id)
|
||||
|
||||
art_lines = [r for r in rows if r[0] > 0]
|
||||
disc_lines = [r for r in rows if r[0] < 0]
|
||||
print(f" Article lines: {len(art_lines)}, Discount lines: {len(disc_lines)}")
|
||||
assert len(art_lines) >= 1, "No article line found"
|
||||
assert len(disc_lines) >= 1, "No discount line found — kit pricing did not activate"
|
||||
print(" PASS")
|
||||
|
||||
conn.commit()
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" FAIL: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def test_repackaging_distributed_total_matches_web():
|
||||
"""Regression (61ae58e): distributed mode total must match web price exactly."""
|
||||
print("\n" + "=" * 60)
|
||||
print("TEST: Repackaging distributed total matches web")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
suffix = f'{datetime.now().strftime("%H%M%S")}-{random.randint(1000, 9999)}'
|
||||
setup_test_data(cur)
|
||||
partner_id = _create_test_partner(cur, suffix)
|
||||
|
||||
# 3 packs @ 400 lei => total web = 1200
|
||||
articles_json = '[{"sku": "CAFE100", "cantitate": 3, "pret": 400}]'
|
||||
order_id = _import_order(cur, f'TEST-BIZ-DIST-{suffix}', partner_id,
|
||||
articles_json, kit_mode='distributed')
|
||||
|
||||
assert order_id and order_id > 0, "Order import failed"
|
||||
rows = _get_order_lines(cur, order_id)
|
||||
|
||||
# Distributed: single line with adjusted price
|
||||
positive_lines = [r for r in rows if r[0] > 0]
|
||||
assert len(positive_lines) == 1, f"Expected 1 line in distributed mode, got {len(positive_lines)}"
|
||||
|
||||
total = positive_lines[0][0] * positive_lines[0][1]
|
||||
expected = 1200.0
|
||||
print(f" Line: qty={positive_lines[0][0]}, price={positive_lines[0][1]:.2f}")
|
||||
print(f" Total: {total:.2f} (expected: {expected:.2f})")
|
||||
assert abs(total - expected) < 0.02, f"Total {total:.2f} != expected {expected:.2f}"
|
||||
print(" PASS")
|
||||
|
||||
conn.commit()
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" FAIL: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def test_kit_markup_no_negative_discount():
|
||||
"""Regression (47b5723): when web price > list price (markup), no discount line should be inserted."""
|
||||
print("\n" + "=" * 60)
|
||||
print("TEST: Kit markup — no negative discount")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
suffix = f'{datetime.now().strftime("%H%M%S")}-{random.randint(1000, 9999)}'
|
||||
setup_test_data(cur)
|
||||
partner_id = _create_test_partner(cur, suffix)
|
||||
|
||||
# CAF01 list price = 51.50/unit, 10 units = 515
|
||||
# Web price 600 > 515 => markup, no discount line
|
||||
articles_json = '[{"sku": "CAFE100", "cantitate": 1, "pret": 600}]'
|
||||
order_id = _import_order(cur, f'TEST-BIZ-MARKUP-{suffix}', partner_id, articles_json)
|
||||
|
||||
assert order_id and order_id > 0, "Order import failed"
|
||||
rows = _get_order_lines(cur, order_id)
|
||||
|
||||
disc_lines = [r for r in rows if r[0] < 0]
|
||||
print(f" Total lines: {len(rows)}, Discount lines: {len(disc_lines)}")
|
||||
assert len(disc_lines) == 0, f"Expected 0 discount lines for markup, got {len(disc_lines)}"
|
||||
print(" PASS")
|
||||
|
||||
conn.commit()
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" FAIL: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def test_kit_component_price_zero_import():
|
||||
"""Regression (1703232): kit components with pret=0 should import successfully."""
|
||||
print("\n" + "=" * 60)
|
||||
print("TEST: Kit component price=0 import")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
suffix = f'{datetime.now().strftime("%H%M%S")}-{random.randint(1000, 9999)}'
|
||||
setup_test_data(cur)
|
||||
partner_id = _create_test_partner(cur, suffix)
|
||||
|
||||
# Temporarily set CAF01 price to 0
|
||||
cur.execute("""
|
||||
UPDATE crm_politici_pret_art SET PRET = 0
|
||||
WHERE id_articol = 9999001 AND id_pol = 1
|
||||
""")
|
||||
conn.commit()
|
||||
|
||||
try:
|
||||
# Import with pret=0 — should succeed (discount = full web price)
|
||||
articles_json = '[{"sku": "CAFE100", "cantitate": 1, "pret": 100}]'
|
||||
order_id = _import_order(cur, f'TEST-BIZ-PRET0-{suffix}', partner_id, articles_json)
|
||||
|
||||
print(f" Order ID: {order_id}")
|
||||
assert order_id and order_id > 0, "Order import failed with pret=0"
|
||||
print(" PASS: Order imported successfully with pret=0")
|
||||
|
||||
conn.commit()
|
||||
finally:
|
||||
# Restore original price
|
||||
cur.execute("""
|
||||
UPDATE crm_politici_pret_art SET PRET = 51.50
|
||||
WHERE id_articol = 9999001 AND id_pol = 1
|
||||
""")
|
||||
conn.commit()
|
||||
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" FAIL: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
# Restore price on error
|
||||
cur.execute("""
|
||||
UPDATE crm_politici_pret_art SET PRET = 51.50
|
||||
WHERE id_articol = 9999001 AND id_pol = 1
|
||||
""")
|
||||
conn.commit()
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def test_duplicate_codmat_different_prices():
|
||||
"""Regression (95565af): same CODMAT at different prices should create separate lines,
|
||||
discriminated by PRET + SIGN(CANTITATE)."""
|
||||
print("\n" + "=" * 60)
|
||||
print("TEST: Duplicate CODMAT different prices")
|
||||
print("=" * 60)
|
||||
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
suffix = f'{datetime.now().strftime("%H%M%S")}-{random.randint(1000, 9999)}'
|
||||
setup_test_data(cur)
|
||||
partner_id = _create_test_partner(cur, suffix)
|
||||
|
||||
# Two articles both mapping to CAF01 but at different prices
|
||||
# CAFE100 -> CAF01 via ARTICOLE_TERTI (kit pricing)
|
||||
# We use separate_line mode so article gets list price 51.50
|
||||
# Then a second article at a different price on the same CODMAT
|
||||
# For this test, we import 2 separate orders to same CODMAT with different prices
|
||||
# The real scenario: kit article line + discount line on same id_articol
|
||||
|
||||
articles_json = '[{"sku": "CAFE100", "cantitate": 1, "pret": 450}]'
|
||||
order_id = _import_order(cur, f'TEST-BIZ-DUP-{suffix}', partner_id, articles_json)
|
||||
|
||||
assert order_id and order_id > 0, "Order import failed"
|
||||
rows = _get_order_lines(cur, order_id)
|
||||
|
||||
# separate_line mode: article at list price + discount at negative qty
|
||||
# Both reference same CODMAT (CAF01) but different PRET and SIGN(CANTITATE)
|
||||
codmats = [r[2] for r in rows]
|
||||
print(f" Lines: {len(rows)}")
|
||||
for r in rows:
|
||||
print(f" qty={r[0]}, pret={r[1]:.2f}, codmat={r[2]}")
|
||||
|
||||
# Should have at least 2 lines with same CODMAT but different qty sign
|
||||
caf_lines = [r for r in rows if r[2] == 'CAF01']
|
||||
assert len(caf_lines) >= 2, f"Expected 2+ CAF01 lines (article + discount), got {len(caf_lines)}"
|
||||
signs = {1 if r[0] > 0 else -1 for r in caf_lines}
|
||||
assert len(signs) == 2, "Expected both positive and negative quantity lines for same CODMAT"
|
||||
print(" PASS: Same CODMAT with different PRET/SIGN coexist")
|
||||
|
||||
conn.commit()
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f" FAIL: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
try:
|
||||
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
teardown_test_data(cur)
|
||||
conn.commit()
|
||||
except:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Starting complete order import test...")
|
||||
print(f"Timestamp: {datetime.now()}")
|
||||
@@ -536,16 +909,32 @@ if __name__ == "__main__":
|
||||
|
||||
print(f"\nTest completed at: {datetime.now()}")
|
||||
if success:
|
||||
print("🎯 PHASE 1 VALIDATION: SUCCESSFUL")
|
||||
print("PHASE 1 VALIDATION: SUCCESSFUL")
|
||||
else:
|
||||
print("🔧 PHASE 1 VALIDATION: NEEDS ATTENTION")
|
||||
print("PHASE 1 VALIDATION: NEEDS ATTENTION")
|
||||
|
||||
# Run repackaging kit pricing test
|
||||
print("\n")
|
||||
repack_success = test_repackaging_kit_pricing()
|
||||
if repack_success:
|
||||
print("🎯 REPACKAGING KIT PRICING: SUCCESSFUL")
|
||||
print("REPACKAGING KIT PRICING: SUCCESSFUL")
|
||||
else:
|
||||
print("🔧 REPACKAGING KIT PRICING: NEEDS ATTENTION")
|
||||
|
||||
print("REPACKAGING KIT PRICING: NEEDS ATTENTION")
|
||||
|
||||
# Run business rule regression tests
|
||||
print("\n")
|
||||
biz_tests = [
|
||||
("Multi-kit discount merge", test_multi_kit_discount_merge),
|
||||
("Kit discount per-kit placement", test_kit_discount_per_kit_placement),
|
||||
("Distributed total matches web", test_repackaging_distributed_total_matches_web),
|
||||
("Markup no negative discount", test_kit_markup_no_negative_discount),
|
||||
("Component price=0 import", test_kit_component_price_zero_import),
|
||||
("Duplicate CODMAT different prices", test_duplicate_codmat_different_prices),
|
||||
]
|
||||
biz_passed = 0
|
||||
for name, test_fn in biz_tests:
|
||||
if test_fn():
|
||||
biz_passed += 1
|
||||
print(f"\nBusiness rule tests: {biz_passed}/{len(biz_tests)} passed")
|
||||
|
||||
exit(0 if success else 1)
|
||||
Reference in New Issue
Block a user