zlwuandCursor
5a9abfdd5d
Persist submitted brand qtys even if plan brand_mode lookup fails; resolve soft-deleted plan products for brand metadata instead of substituting another product; stop seeding all actual qty onto the first brand option. Co-authored-by: Cursor <cursoragent@cursor.com>
129 lines
3.3 KiB
Python
129 lines
3.3 KiB
Python
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from app.services.pl_feedback_brand_qty import format_brand_qty_summary, require_brand_qtys_if_needed
|
|
|
|
|
|
def test_format_brand_qty_summary():
|
|
s = format_brand_qty_summary([
|
|
{"brandName": "君正", "qty": Decimal("16")},
|
|
{"brandName": "红四方", "qty": Decimal("12.000")},
|
|
])
|
|
assert s == "君正 16 / 红四方 12"
|
|
|
|
|
|
def test_format_brand_qty_summary_empty():
|
|
assert format_brand_qty_summary([]) is None
|
|
|
|
|
|
def test_format_brand_qty_summary_zero_only():
|
|
assert format_brand_qty_summary([{"brandName": "君正", "qty": Decimal("0")}]) is None
|
|
|
|
|
|
def test_require_skips_when_no_brand_mode():
|
|
require_brand_qtys_if_needed(
|
|
brand_mode=None,
|
|
required_brand_id=None,
|
|
actual_qty=Decimal("10"),
|
|
rows=[],
|
|
enforce=True,
|
|
)
|
|
|
|
|
|
def test_require_skips_when_not_enforced():
|
|
require_brand_qtys_if_needed(
|
|
brand_mode="any",
|
|
required_brand_id=None,
|
|
actual_qty=Decimal("10"),
|
|
rows=[],
|
|
enforce=False,
|
|
)
|
|
|
|
|
|
def test_require_enforces_when_completed():
|
|
with pytest.raises(ValueError, match="至少一个品牌数量"):
|
|
require_brand_qtys_if_needed(
|
|
brand_mode="any",
|
|
required_brand_id=None,
|
|
actual_qty=Decimal("10"),
|
|
rows=[],
|
|
enforce=True,
|
|
)
|
|
|
|
|
|
def test_require_calls_validate_when_positive_rows(monkeypatch):
|
|
called = {}
|
|
|
|
def mock_validate(**kwargs):
|
|
called.update(kwargs)
|
|
|
|
monkeypatch.setattr(
|
|
"app.services.pl_feedback_brand_qty.validate_brand_qty_rows",
|
|
mock_validate,
|
|
)
|
|
rows = [{"brandId": "b1", "brandName": "君正", "qty": Decimal("10")}]
|
|
require_brand_qtys_if_needed(
|
|
brand_mode="any",
|
|
required_brand_id=None,
|
|
actual_qty=Decimal("10"),
|
|
rows=rows,
|
|
enforce=True,
|
|
)
|
|
assert called["brand_mode"] == "any"
|
|
assert called["actual_qty"] == Decimal("10")
|
|
assert called["rows"] == rows
|
|
|
|
|
|
def test_persist_io_item_brand_qtys_writes_when_brand_mode_missing():
|
|
"""计划行 brand_mode 丢失时，客户端提交的多品牌明细仍应落库（不能只软删不写）。"""
|
|
from app.services.pl_feedback_brand_qty import persist_io_item_brand_qtys
|
|
|
|
class _Row:
|
|
def __init__(self):
|
|
self.del_flag = 0
|
|
self.update_by = None
|
|
|
|
existing = [_Row()]
|
|
added = []
|
|
|
|
class _Query:
|
|
def filter(self, *args, **kwargs):
|
|
return self
|
|
|
|
def all(self):
|
|
return existing
|
|
|
|
class _Db:
|
|
def query(self, model):
|
|
return _Query()
|
|
|
|
def add(self, obj):
|
|
added.append(obj)
|
|
|
|
class _Model:
|
|
order_item_id = object()
|
|
tenant_no = object()
|
|
del_flag = object()
|
|
|
|
def __init__(self, **kwargs):
|
|
self.__dict__.update(kwargs)
|
|
|
|
persist_io_item_brand_qtys(
|
|
_Db(),
|
|
model=_Model,
|
|
tenant_no="t1",
|
|
order_id="o1",
|
|
order_item_id="i1",
|
|
plan_product_id="p1",
|
|
brand_mode=None,
|
|
rows=[
|
|
{"brandId": "b1", "brandName": "君正", "qty": Decimal("20")},
|
|
{"brandId": "b2", "brandName": "红四方", "qty": Decimal("30")},
|
|
],
|
|
user_id="u1",
|
|
)
|
|
assert existing[0].del_flag == 1
|
|
assert len(added) == 2
|
|
assert {a.brand_id for a in added} == {"b1", "b2"}
|