diff --git a/modules/purchase_trade/duplicate.py b/modules/purchase_trade/duplicate.py index 70dea94..7e7d9a2 100644 --- a/modules/purchase_trade/duplicate.py +++ b/modules/purchase_trade/duplicate.py @@ -3,7 +3,7 @@ from decimal import Decimal from types import SimpleNamespace -from trytond.exceptions import UserError +from trytond.exceptions import UserError, UserWarning from trytond.model import ModelView, fields from trytond.pool import Pool from trytond.pyson import Bool, Eval @@ -110,7 +110,8 @@ class TradeCustomDuplicate(Wizard): line = self._trade_line(record) if not line: raise UserError("The selected contract has no trade line to duplicate.") - matched_record = self._matched_counterpart_record(line, strict=False) + matched_record = self._matched_counterpart_record( + record, line, strict=False) mirror_mode = 'duplicate_pair' if matched_record else 'new_mirror' return { @@ -335,7 +336,7 @@ class TradeCustomDuplicate(Wizard): return records[0] if records else None @staticmethod - def _matched_counterpart_line(line, strict=True): + def _matched_counterpart_line(line, strict=True, warn=False): side = TradeCustomDuplicate._line_side(line) if side not in {'purchase', 'sale'}: if strict: @@ -360,6 +361,10 @@ class TradeCustomDuplicate(Wizard): lines = [ lqt.lot_s.sale_line for lqt in lot_qts if getattr(getattr(lqt, 'lot_s', None), 'sale_line', None)] + lines.extend( + lot.sale_line for lot in (getattr(line, 'lots', None) or []) + if (getattr(lot, 'lot_type', None) == 'physic' + and getattr(lot, 'sale_line', None))) else: lot_qts = LotQt.search([ ('lot_s', 'in', lot_ids), @@ -369,6 +374,10 @@ class TradeCustomDuplicate(Wizard): lines = [ lqt.lot_p.line for lqt in lot_qts if getattr(getattr(lqt, 'lot_p', None), 'line', None)] + lines.extend( + lot.line for lot in (getattr(line, 'lots', None) or []) + if (getattr(lot, 'lot_type', None) == 'physic' + and getattr(lot, 'line', None))) unique = [] seen = set() @@ -379,24 +388,98 @@ class TradeCustomDuplicate(Wizard): seen.add(matched_id) unique.append(matched_line) - if len(unique) == 1: + if unique: + if warn and len(unique) > 1: + TradeCustomDuplicate._warn_multiple_counterparts(line) return unique[0] if strict: raise UserError( - "Duplicate matched pair requires exactly one matched " + "Duplicate matched pair requires an existing matched " "counterpart.") return None @staticmethod - def _matched_counterpart_record(line, strict=True): - matched_line = TradeCustomDuplicate._matched_counterpart_line( - line, strict=strict) - if not matched_line: - return None + def _warn_multiple_counterparts(line): + Warning = Pool().get('res.user.warning') + warning_name = 'custom_duplicate_multiple_counterparts:%s' % ( + getattr(line, 'id', id(line))) + if Warning.check(warning_name): + raise UserWarning( + warning_name, + "Several matched counterpart contracts were found. The custom " + "duplicate will use the first one by default. Continue only " + "if that is the expected sale/purchase to duplicate.") + + @staticmethod + def _counterpart_record_from_line(matched_line): return ( getattr(matched_line, 'sale', None) or getattr(matched_line, 'purchase', None)) + @staticmethod + def _btb_counterpart_line(record, strict=True, warn=False): + btb = getattr(record, 'btb', None) + if not btb: + if strict: + raise UserError( + "Duplicate matched pair requires an existing matched " + "lot or back-to-back counterpart.") + return None + + source_model = TradeCustomDuplicate._source_model(record) + counterpart_records = ( + getattr(btb, 'sale', None) + if source_model == 'purchase.purchase' + else getattr(btb, 'purchase', None)) + lines = [] + for counterpart in counterpart_records or []: + line = TradeCustomDuplicate._trade_line(counterpart) + if line: + lines.append(line) + + unique = [] + seen = set() + for line in lines: + line_id = getattr(line, 'id', id(line)) + if line_id in seen: + continue + seen.add(line_id) + unique.append(line) + + if unique: + if warn and len(unique) > 1: + TradeCustomDuplicate._warn_multiple_counterparts( + TradeCustomDuplicate._trade_line(record)) + return unique[0] + if strict: + raise UserError( + "Duplicate matched pair requires an existing matched " + "counterpart.") + return None + + @staticmethod + def _matched_counterpart(line, record=None, strict=True): + matched_line = TradeCustomDuplicate._matched_counterpart_line( + line, strict=False, warn=strict) + if not matched_line and record: + matched_line = TradeCustomDuplicate._btb_counterpart_line( + record, strict=False, warn=strict) + if not matched_line: + if strict: + raise UserError( + "Duplicate matched pair requires exactly one matched " + "counterpart.") + return None, None + return ( + matched_line, + TradeCustomDuplicate._counterpart_record_from_line(matched_line)) + + @staticmethod + def _matched_counterpart_record(record, line, strict=True): + matched_line, matched_record = TradeCustomDuplicate._matched_counterpart( + line, record=record, strict=strict) + return matched_record + @staticmethod def _pair_duplicate_options(matched_record, matched_line, options): return SimpleNamespace( @@ -451,10 +534,8 @@ class TradeCustomDuplicate(Wizard): @staticmethod def _duplicate_matched_pair(record, line, new_line, options): - matched_line = TradeCustomDuplicate._matched_counterpart_line(line) - matched_record = ( - getattr(matched_line, 'sale', None) - or getattr(matched_line, 'purchase', None)) + matched_line, matched_record = TradeCustomDuplicate._matched_counterpart( + line, record=record) if not matched_record: raise UserError( "Duplicate matched pair could not find the matched contract.") diff --git a/modules/purchase_trade/tests/test_module.py b/modules/purchase_trade/tests/test_module.py index 99b6b36..ddc04de 100644 --- a/modules/purchase_trade/tests/test_module.py +++ b/modules/purchase_trade/tests/test_module.py @@ -6249,6 +6249,160 @@ class PurchaseTradeTestCase(ModuleTestCase): self.assertEqual(values['mirror_mode'], 'duplicate_pair') self.assertEqual(values['mirror_party'], sale_party.id) + def test_custom_duplicate_defaults_mirror_from_back_to_back_sale(self): + 'custom duplicate can use the back-to-back sale as matched pair' + purchase_party = SimpleNamespace(id=1) + sale_party = SimpleNamespace(id=2) + sale_line = SimpleNamespace( + __name__='sale.line', + id=20, + type='line', + product=Mock()) + sale = SimpleNamespace( + __name__='sale.sale', + id=21, + lines=[sale_line], + party=sale_party) + sale_line.sale = sale + btb = SimpleNamespace(sale=[sale], purchase=[]) + + class FakeLotQt: + @classmethod + def search(cls, domain): + return [] + + pool = Mock() + pool.get.side_effect = lambda name: {'lot.qt': FakeLotQt}[name] + line = SimpleNamespace( + __name__='purchase.line', + id=11, + type='line', + product=Mock(), + lots=[], + unit=SimpleNamespace(id=3), + quantity=Decimal('100'), + quantity_theorical=Decimal('100'), + unit_price=Decimal('25'), + price_type='priced', + ) + record = SimpleNamespace( + __name__='purchase.purchase', + id=12, + lines=[line], + party=purchase_party, + currency=SimpleNamespace(id=4), + payment_term=SimpleNamespace(id=5), + incoterm=SimpleNamespace(id=6), + btb=btb, + ) + wizard = duplicate_module.TradeCustomDuplicate() + wizard.records = [record] + + with patch.object(duplicate_module, 'Pool', return_value=pool): + values = wizard.default_start(None) + + self.assertEqual(values['mirror_mode'], 'duplicate_pair') + self.assertEqual(values['mirror_party'], sale_party.id) + + def test_custom_duplicate_defaults_mirror_from_physical_lot_sale(self): + 'custom duplicate can use the sale_line of a physical lot' + purchase_party = SimpleNamespace(id=1) + sale_party = SimpleNamespace(id=2) + sale = SimpleNamespace(party=sale_party) + sale_line = SimpleNamespace(id=20, sale=sale) + physical_lot = SimpleNamespace( + id=10, + lot_type='physic', + sale_line=sale_line) + + class FakeLotQt: + @classmethod + def search(cls, domain): + return [] + + pool = Mock() + pool.get.side_effect = lambda name: {'lot.qt': FakeLotQt}[name] + line = SimpleNamespace( + __name__='purchase.line', + id=11, + type='line', + product=Mock(), + lots=[physical_lot], + unit=SimpleNamespace(id=3), + quantity=Decimal('100'), + quantity_theorical=Decimal('100'), + unit_price=Decimal('25'), + price_type='priced', + ) + record = SimpleNamespace( + __name__='purchase.purchase', + id=12, + lines=[line], + party=purchase_party, + currency=SimpleNamespace(id=4), + payment_term=SimpleNamespace(id=5), + incoterm=SimpleNamespace(id=6), + ) + wizard = duplicate_module.TradeCustomDuplicate() + wizard.records = [record] + + with patch.object(duplicate_module, 'Pool', return_value=pool): + values = wizard.default_start(None) + + self.assertEqual(values['mirror_mode'], 'duplicate_pair') + self.assertEqual(values['mirror_party'], sale_party.id) + + def test_custom_duplicate_warns_and_uses_first_counterpart(self): + 'custom duplicate warns when several matched counterparts exist' + sale_line_a = SimpleNamespace(id=20, sale=SimpleNamespace(id=30)) + sale_line_b = SimpleNamespace(id=21, sale=SimpleNamespace(id=31)) + lot_qts = [ + SimpleNamespace(lot_s=SimpleNamespace(sale_line=sale_line_a)), + SimpleNamespace(lot_s=SimpleNamespace(sale_line=sale_line_b)), + ] + + class FakeLotQt: + @classmethod + def search(cls, domain): + return lot_qts + + class FakeWarning: + @classmethod + def check(cls, name): + return True + + pool = Mock() + pool.get.side_effect = lambda name: { + 'lot.qt': FakeLotQt, + 'res.user.warning': FakeWarning, + }[name] + line = SimpleNamespace( + __name__='purchase.line', + id=11, + lots=[SimpleNamespace(id=10)], + ) + + with patch.object(duplicate_module, 'Pool', return_value=pool): + with self.assertRaises(UserWarning): + duplicate_module.TradeCustomDuplicate._matched_counterpart_line( + line, strict=False, warn=True) + + class AcknowledgedWarning: + @classmethod + def check(cls, name): + return False + + pool.get.side_effect = lambda name: { + 'lot.qt': FakeLotQt, + 'res.user.warning': AcknowledgedWarning, + }[name] + with patch.object(duplicate_module, 'Pool', return_value=pool): + matched_line = ( + duplicate_module.TradeCustomDuplicate._matched_counterpart_line( + line, strict=False, warn=True)) + + self.assertIs(matched_line, sale_line_a) + def test_custom_duplicate_match_pair_lots_links_new_quantities(self): 'custom duplicate matches the new purchase and sale lot.qt rows' saved = []