diff --git a/modules/purchase_trade/duplicate.py b/modules/purchase_trade/duplicate.py
index f62e9f1..894b7df 100644
--- a/modules/purchase_trade/duplicate.py
+++ b/modules/purchase_trade/duplicate.py
@@ -1,6 +1,7 @@
# This file is part of Tryton. The COPYRIGHT file at the top level of
# this repository contains the full copyright notices and license terms.
from decimal import Decimal
+from types import SimpleNamespace
from trytond.exceptions import UserError
from trytond.model import ModelView, fields
@@ -19,6 +20,11 @@ PRICE_TYPES = [
('efp', 'EFP'),
]
+MIRROR_MODES = [
+ ('duplicate_pair', 'Duplicate matched pair'),
+ ('new_mirror', 'Create new mirror contract'),
+]
+
class TradeCustomDuplicateStart(ModelView):
"Custom Duplicate"
@@ -46,6 +52,11 @@ class TradeCustomDuplicateStart(ModelView):
clear_fees = fields.Boolean("Clear Fees")
clear_pricing_components = fields.Boolean("Clear Pricing Components")
create_matched_mirror = fields.Boolean("Create matched mirror contract")
+ mirror_mode = fields.Selection(MIRROR_MODES, "Mirror Mode",
+ states={
+ 'invisible': ~Bool(Eval('create_matched_mirror')),
+ 'required': Bool(Eval('create_matched_mirror')),
+ })
mirror_party = fields.Many2One('party.party', "Mirror Counterparty",
states={
'invisible': ~Bool(Eval('create_matched_mirror')),
@@ -97,6 +108,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)
+ mirror_mode = 'duplicate_pair' if matched_record else 'new_mirror'
return {
'source_model': source_model,
@@ -115,6 +128,10 @@ class TradeCustomDuplicate(Wizard):
'clear_fees': False,
'clear_pricing_components': False,
'create_matched_mirror': False,
+ 'mirror_mode': mirror_mode,
+ 'mirror_party': (
+ matched_record.party.id
+ if matched_record and matched_record.party else None),
}
@staticmethod
@@ -293,6 +310,168 @@ class TradeCustomDuplicate(Wizard):
if not LotQt.search(domain):
create()
+ @staticmethod
+ def _open_lot_qt(lot, side):
+ LotQt = Pool().get('lot.qt')
+ if side == 'purchase':
+ domain = [
+ ('lot_p', '=', lot.id),
+ ('lot_s', '=', None),
+ ('lot_shipment_in', '=', None),
+ ('lot_shipment_internal', '=', None),
+ ('lot_shipment_out', '=', None),
+ ]
+ else:
+ domain = [
+ ('lot_p', '=', None),
+ ('lot_s', '=', lot.id),
+ ('lot_shipment_in', '=', None),
+ ('lot_shipment_internal', '=', None),
+ ('lot_shipment_out', '=', None),
+ ]
+ records = LotQt.search(domain)
+ return records[0] if records else None
+
+ @staticmethod
+ def _matched_counterpart_line(line, strict=True):
+ side = TradeCustomDuplicate._line_side(line)
+ if side not in {'purchase', 'sale'}:
+ if strict:
+ raise UserError(
+ "Custom duplicate cannot determine the contract side.")
+ return None
+
+ lot_ids = [lot.id for lot in (getattr(line, 'lots', None) or [])]
+ if not lot_ids:
+ if strict:
+ raise UserError(
+ "Duplicate matched pair requires an existing matched lot.")
+ return None
+
+ LotQt = Pool().get('lot.qt')
+ if side == 'purchase':
+ lot_qts = LotQt.search([
+ ('lot_p', 'in', lot_ids),
+ ('lot_s', '!=', None),
+ ('lot_quantity', '>', 0),
+ ])
+ lines = [
+ lqt.lot_s.sale_line for lqt in lot_qts
+ if getattr(getattr(lqt, 'lot_s', None), 'sale_line', None)]
+ else:
+ lot_qts = LotQt.search([
+ ('lot_s', 'in', lot_ids),
+ ('lot_p', '!=', None),
+ ('lot_quantity', '>', 0),
+ ])
+ lines = [
+ lqt.lot_p.line for lqt in lot_qts
+ if getattr(getattr(lqt, 'lot_p', None), 'line', None)]
+
+ unique = []
+ seen = set()
+ for matched_line in lines:
+ matched_id = getattr(matched_line, 'id', id(matched_line))
+ if matched_id in seen:
+ continue
+ seen.add(matched_id)
+ unique.append(matched_line)
+
+ if len(unique) == 1:
+ return unique[0]
+ if strict:
+ raise UserError(
+ "Duplicate matched pair requires exactly one 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
+ return (
+ getattr(matched_line, 'sale', None)
+ or getattr(matched_line, 'purchase', None))
+
+ @staticmethod
+ def _pair_duplicate_options(matched_record, matched_line, options):
+ return SimpleNamespace(
+ party=options.mirror_party or matched_record.party,
+ currency=matched_record.currency or options.currency,
+ payment_term=(
+ matched_record.payment_term or options.payment_term),
+ incoterm=matched_record.incoterm or options.incoterm,
+ quantity=options.quantity,
+ unit_price=getattr(matched_line, 'unit_price', None),
+ price_type=(
+ getattr(matched_line, 'price_type', None)
+ or options.price_type),
+ clear_fees=options.clear_fees,
+ clear_pricing_components=options.clear_pricing_components,
+ )
+
+ @staticmethod
+ def _match_duplicate_pair_lots(source_line, source_lot, counterpart_line,
+ counterpart_lot, quantity):
+ LotQt = Pool().get('lot.qt')
+ Lot = Pool().get('lot.lot')
+ source_side = TradeCustomDuplicate._line_side(source_line)
+ if source_side not in {'purchase', 'sale'}:
+ raise UserError(
+ "Custom duplicate cannot determine the contract side.")
+ if source_side == 'purchase':
+ purchase_line = source_line
+ purchase_lot = source_lot
+ sale_line = counterpart_line
+ sale_lot = counterpart_lot
+ else:
+ purchase_line = counterpart_line
+ purchase_lot = counterpart_lot
+ sale_line = source_line
+ sale_lot = source_lot
+
+ purchase_lqt = TradeCustomDuplicate._open_lot_qt(
+ purchase_lot, 'purchase')
+ sale_lqt = TradeCustomDuplicate._open_lot_qt(sale_lot, 'sale')
+ if not purchase_lqt or not sale_lqt:
+ raise UserError(
+ "Duplicate matched pair could not find the new open lot "
+ "quantities to match.")
+
+ purchase_lqt.lot_s = sale_lot.id
+ purchase_lqt.lot_quantity = Decimal(str(quantity)).quantize(
+ Decimal("0.00001"))
+ LotQt.save([purchase_lqt])
+ LotQt.delete([sale_lqt])
+ Lot.assert_lines_quantity_consistency([purchase_line, sale_line])
+
+ @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))
+ if not matched_record:
+ raise UserError(
+ "Duplicate matched pair could not find the matched contract.")
+
+ mirror_options = TradeCustomDuplicate._pair_duplicate_options(
+ matched_record, matched_line, options)
+ new_matched_record = TradeCustomDuplicate._copy_contract(
+ matched_record, mirror_options)
+ new_matched_line = TradeCustomDuplicate._apply_line_options(
+ new_matched_record, mirror_options)
+ source_lot = TradeCustomDuplicate._ensure_open_virtual_lot(new_line)
+ counterpart_lot = TradeCustomDuplicate._ensure_open_virtual_lot(
+ new_matched_line)
+ if source_lot and counterpart_lot:
+ TradeCustomDuplicate._match_duplicate_pair_lots(
+ new_line, source_lot, new_matched_line, counterpart_lot,
+ options.quantity)
+ return new_matched_record
+
@staticmethod
def _create_matched_mirror(record, line, options):
source_model = record.__name__
@@ -348,7 +527,11 @@ class TradeCustomDuplicate(Wizard):
TradeCustomDuplicate._ensure_open_virtual_lot(new_line)
self.start.duplicated_record_id = new_record.id
if options.create_matched_mirror:
- self._create_matched_mirror(new_record, new_line, options)
+ if options.mirror_mode == 'duplicate_pair':
+ self._duplicate_matched_pair(record, self._trade_line(record),
+ new_line, options)
+ else:
+ self._create_matched_mirror(new_record, new_line, options)
if new_record.__name__ == 'purchase.purchase':
return 'open_purchase'
return 'open_sale'
diff --git a/modules/purchase_trade/tests/test_module.py b/modules/purchase_trade/tests/test_module.py
index 7c7eccb..99b6b36 100644
--- a/modules/purchase_trade/tests/test_module.py
+++ b/modules/purchase_trade/tests/test_module.py
@@ -6203,6 +6203,108 @@ class PurchaseTradeTestCase(ModuleTestCase):
self.assertEqual(virtual_parts, [
(43, Decimal('50.00000'), None, 43, 'only sale')])
+ def test_custom_duplicate_defaults_mirror_from_matched_sale(self):
+ 'custom duplicate can default mirror party from the matched sale'
+ purchase_party = SimpleNamespace(id=1)
+ sale_party = SimpleNamespace(id=2)
+ purchase_lot = SimpleNamespace(id=10)
+ sale = SimpleNamespace(party=sale_party)
+ sale_line = SimpleNamespace(id=20, sale=sale)
+ lot_qt = SimpleNamespace(lot_s=SimpleNamespace(sale_line=sale_line))
+
+ class FakeLotQt:
+ @classmethod
+ def search(cls, domain):
+ return [lot_qt]
+
+ pool = Mock()
+ pool.get.side_effect = lambda name: {'lot.qt': FakeLotQt}[name]
+ line = SimpleNamespace(
+ __name__='purchase.line',
+ id=11,
+ type='line',
+ product=Mock(),
+ lots=[purchase_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_match_pair_lots_links_new_quantities(self):
+ 'custom duplicate matches the new purchase and sale lot.qt rows'
+ saved = []
+ deleted = []
+ checked_lines = []
+ purchase_lqt = SimpleNamespace(
+ lot_s=None,
+ lot_quantity=Decimal('0'),
+ )
+ sale_lqt = SimpleNamespace(
+ lot_p=None,
+ lot_quantity=Decimal('100'),
+ )
+
+ class FakeLotQt:
+ @classmethod
+ def search(cls, domain):
+ if ('lot_p', '=', 101) in domain:
+ return [purchase_lqt]
+ if ('lot_s', '=', 202) in domain:
+ return [sale_lqt]
+ return []
+
+ @classmethod
+ def save(cls, records):
+ saved.extend(records)
+
+ @classmethod
+ def delete(cls, records):
+ deleted.extend(records)
+
+ class FakeLot:
+ @classmethod
+ def assert_lines_quantity_consistency(cls, lines):
+ checked_lines.extend(lines)
+
+ pool = Mock()
+ pool.get.side_effect = lambda name: {
+ 'lot.qt': FakeLotQt,
+ 'lot.lot': FakeLot,
+ }[name]
+ purchase_line = SimpleNamespace(__name__='purchase.line')
+ sale_line = SimpleNamespace(__name__='sale.line')
+
+ with patch.object(duplicate_module, 'Pool', return_value=pool):
+ duplicate_module.TradeCustomDuplicate._match_duplicate_pair_lots(
+ purchase_line, SimpleNamespace(id=101),
+ sale_line, SimpleNamespace(id=202),
+ Decimal('100'))
+
+ self.assertEqual(purchase_lqt.lot_s, 202)
+ self.assertEqual(purchase_lqt.lot_quantity, Decimal('100.00000'))
+ self.assertEqual(saved, [purchase_lqt])
+ self.assertEqual(deleted, [sale_lqt])
+ self.assertEqual(checked_lines, [purchase_line, sale_line])
+
def test_lot_matching_rejects_purchase_quantity_above_available(self):
'apply matching cannot consume more than available purchase quantity'
class LotQtMock:
diff --git a/modules/purchase_trade/view/trade_custom_duplicate_start_form.xml b/modules/purchase_trade/view/trade_custom_duplicate_start_form.xml
index a0f827c..e74d4fc 100644
--- a/modules/purchase_trade/view/trade_custom_duplicate_start_form.xml
+++ b/modules/purchase_trade/view/trade_custom_duplicate_start_form.xml
@@ -24,6 +24,8 @@
+
+