Th qt modification

modules/purchase_trade/docs/business-rules.md

@@ -566,6 +566,30 @@ Owner technique: `a completer`
 - Priorite:
   - `importante`
 
+### BR-PT-020 - Le solde `lot.qt` ouvert suit les lots physiques existants
+
+- Intent: eviter qu'une hausse ou baisse de quantite contractuelle double le
+  reliquat ouvert quand des lots physiques existent deja.
+- Description:
+  - Lorsqu'une `purchase.line.quantity_theorical` ou
+    `sale.line.quantity_theorical` est modifiee, le `lot.qt` libre ne doit pas
+    etre ajuste uniquement par delta.
+  - Le systeme doit recalculer le solde ouvert cible depuis la quantite
+    contractuelle, les lots physiques deja crees et les quantites deja
+    allouees dans des `lot.qt` matches ou shippes.
+- Resultat attendu:
+  - quantite virtuelle cible =
+    `quantity_theorical - somme(lots physiques convertis dans l'unite ligne)`
+  - `lot.qt` libre non matche / non shippe =
+    `quantite virtuelle cible - somme(lot.qt deja matches ou shippes)`
+  - si le resultat devient negatif, bloquer avec
+    `Please unlink or unmatch lot`
+  - exemple: une ligne achat passee de `10000` a `20000` avec deja `10000`
+    physiques doit afficher `10000` ouverts et `10000` physiques, pas
+    `20000` ouverts plus `10000` physiques.
+- Priorite:
+  - `structurante`
+
 ## 4) Exemples concrets
 
 ### Exemple E1 - Augmentation simple

modules/purchase_trade/purchase.py

@@ -1530,8 +1530,6 @@ class Line(metaclass=PoolMeta):
         # alors il faut créer un nouveau lot_qt non shippé et non matché avec le delta
         # Si delta négatif alors on decrease si c'est possible le lot_qt non shippé non matché et s'il n'y en a pas on envoie un 
         # message d'erreur 'Please unlink or unmatch lot'
-        Lot = Pool().get('lot.lot')
-        LotQt = Pool().get('lot.qt')
         old_values = {}
 
         for records, values in zip(args[::2], args[1::2]):
@@ -1558,46 +1556,63 @@ class Line(metaclass=PoolMeta):
                     else Decimal(vlot.get_current_quantity_converted() or 0)
                 )
                 delta = new - baseline
-                if delta > 0:
+                if delta:
-                    new_qty = round(Decimal(vlot.get_current_quantity_converted() or 0) + delta, 5)
+                    physical_quantity = sum(
-                    vlot.set_current_quantity(new_qty, new_qty, 1)
+                        Decimal(lot.get_current_quantity_converted() or 0)
-                    Lot.save([vlot])
+                        for lot in (line.lots or [])
-                    lqts = LotQt.search([
+                        if lot.lot_type == 'physic')
+                    target_quantity = round(new - physical_quantity, 5)
+                    if target_quantity < 0:
+                        raise UserError("Please unlink or unmatch lot")
+                    cls._sync_open_lot_quantity(line, vlot, target_quantity)
+
+    @classmethod
+    def _sync_open_lot_quantity(cls, line, vlot, target_quantity):
+        Lot = Pool().get('lot.lot')
+        LotQt = Pool().get('lot.qt')
+
+        free_domain = [
             ('lot_p', '=', vlot.id),
             ('lot_s', '=', None),
             ('lot_shipment_in', '=', None),
             ('lot_shipment_internal', '=', None),
             ('lot_shipment_out', '=', None),
+        ]
+        free_lqts = LotQt.search(free_domain)
+        allocated_lqts = LotQt.search([
+            ('lot_p', '=', vlot.id),
+            [
+                'OR',
+                ('lot_s', '!=', None),
+                ('lot_shipment_in', '!=', None),
+                ('lot_shipment_internal', '!=', None),
+                ('lot_shipment_out', '!=', None),
+            ],
         ])
-                    if lqts:
+        allocated_quantity = sum(
-                        lqt = lqts[0]
+            Decimal(lqt.lot_quantity or 0) for lqt in allocated_lqts)
-                        lqt.lot_quantity = round(Decimal(lqt.lot_quantity or 0) + delta, 5)
+        free_quantity = round(target_quantity - allocated_quantity, 5)
+        if free_quantity < 0:
+            raise UserError("Please unlink or unmatch lot")
+
+        current_quantity = round(
+            Decimal(vlot.get_current_quantity_converted() or 0), 5)
+        if current_quantity != target_quantity:
+            vlot.set_current_quantity(target_quantity, target_quantity, 1)
+            Lot.save([vlot])
+
+        if free_lqts:
+            lqt = free_lqts[0]
+            if Decimal(lqt.lot_quantity or 0) != free_quantity:
+                lqt.lot_quantity = free_quantity
                 LotQt.save([lqt])
-                    else:
+        elif free_quantity > 0:
             lqt = LotQt()
             lqt.lot_p = vlot.id
             lqt.lot_s = None
-                        lqt.lot_quantity = round(delta, 5)
+            lqt.lot_quantity = free_quantity
             lqt.lot_unit = line.unit
             LotQt.save([lqt])
-                elif delta < 0:
-                    decrease = abs(delta)
-                    lqts = LotQt.search([
-                        ('lot_p', '=', vlot.id),
-                        ('lot_s', '=', None),
-                        ('lot_shipment_in', '=', None),
-                        ('lot_shipment_internal', '=', None),
-                        ('lot_shipment_out', '=', None),
-                    ])
-                    if (not lqts
-                            or Decimal(lqts[0].lot_quantity or 0) < decrease):
-                        raise UserError("Please unlink or unmatch lot")
-                    new_qty = round(Decimal(vlot.get_current_quantity_converted() or 0) - decrease, 5)
-                    vlot.set_current_quantity(new_qty, new_qty, 1)
-                    Lot.save([vlot])
-                    lqt = lqts[0]
-                    lqt.lot_quantity = round(Decimal(lqt.lot_quantity or 0) - decrease, 5)
-                    LotQt.save([lqt])
 
     @classmethod
     def copy(cls, lines, default=None):

modules/purchase_trade/sale.py

@@ -1644,8 +1644,6 @@ class SaleLine(metaclass=PoolMeta):
             cls._check_delivery_period_values(records, values)
             args.extend((records, values))
 
-        Lot = Pool().get('lot.lot')
-        LotQt = Pool().get('lot.qt')
         old_values = {}
 
         for records, values in zip(args[::2], args[1::2]):
@@ -1675,46 +1673,61 @@ class SaleLine(metaclass=PoolMeta):
                 continue
 
             vlot = virtual_lots[0]
-            lqts = LotQt.search([
+            physical_quantity = sum(
+                Decimal(lot.get_current_quantity_converted() or 0)
+                for lot in (line.lots or [])
+                if lot.lot_type == 'physic')
+            target_quantity = round(new - physical_quantity, 5)
+            if target_quantity < 0:
+                raise UserError("Please unlink or unmatch lot")
+            cls._sync_open_lot_quantity(line, vlot, target_quantity)
+
+    @classmethod
+    def _sync_open_lot_quantity(cls, line, vlot, target_quantity):
+        Lot = Pool().get('lot.lot')
+        LotQt = Pool().get('lot.qt')
+
+        free_lqts = LotQt.search([
             ('lot_s', '=', vlot.id),
             ('lot_p', '=', None),
             ('lot_shipment_in', '=', None),
             ('lot_shipment_internal', '=', None),
             ('lot_shipment_out', '=', None),
         ])
+        allocated_lqts = LotQt.search([
+            ('lot_s', '=', vlot.id),
+            [
+                'OR',
+                ('lot_p', '!=', None),
+                ('lot_shipment_in', '!=', None),
+                ('lot_shipment_internal', '!=', None),
+                ('lot_shipment_out', '!=', None),
+            ],
+        ])
+        allocated_quantity = sum(
+            Decimal(lqt.lot_quantity or 0) for lqt in allocated_lqts)
+        free_quantity = round(target_quantity - allocated_quantity, 5)
+        if free_quantity < 0:
+            raise UserError("Please unlink or unmatch lot")
 
-            if delta > 0:
+        current_quantity = round(
-                new_qty = round(
+            Decimal(vlot.get_current_quantity_converted() or 0), 5)
-                    Decimal(vlot.get_current_quantity_converted() or 0) + delta,
+        if current_quantity != target_quantity:
-                    5)
+            vlot.set_current_quantity(target_quantity, target_quantity, 1)
-                vlot.set_current_quantity(new_qty, new_qty, 1)
             Lot.save([vlot])
-                if lqts:
+
-                    lqt = lqts[0]
+        if free_lqts:
-                    lqt.lot_quantity = round(
+            lqt = free_lqts[0]
-                        Decimal(lqt.lot_quantity or 0) + delta, 5)
+            if Decimal(lqt.lot_quantity or 0) != free_quantity:
+                lqt.lot_quantity = free_quantity
                 LotQt.save([lqt])
-                else:
+        elif free_quantity > 0:
             lqt = LotQt()
             lqt.lot_p = None
             lqt.lot_s = vlot.id
-                    lqt.lot_quantity = round(delta, 5)
+            lqt.lot_quantity = free_quantity
             lqt.lot_unit = line.unit
             LotQt.save([lqt])
-            elif delta < 0:
-                decrease = abs(delta)
-                if not lqts or Decimal(lqts[0].lot_quantity or 0) < decrease:
-                    raise UserError("Please unlink or unmatch lot")
-                new_qty = round(
-                    Decimal(vlot.get_current_quantity_converted() or 0)
-                    - decrease,
-                    5)
-                vlot.set_current_quantity(new_qty, new_qty, 1)
-                Lot.save([vlot])
-                lqt = lqts[0]
-                lqt.lot_quantity = round(
-                    Decimal(lqt.lot_quantity or 0) - decrease, 5)
-                LotQt.save([lqt])
 
     @classmethod
     def delete(cls, lines):

modules/purchase_trade/tests/test_module.py

@@ -1061,7 +1061,7 @@ class PurchaseTradeTestCase(ModuleTestCase):
 
         lot_model = Mock()
         lotqt_model = Mock()
-        lotqt_model.search.return_value = [lqt]
+        lotqt_model.search.side_effect = [[lqt], []]
 
         with patch(
                 'trytond.modules.purchase_trade.sale.Pool'
@@ -1097,10 +1097,11 @@ class PurchaseTradeTestCase(ModuleTestCase):
         vlot.get_current_quantity_converted.return_value = Decimal('10')
         line.lots = [vlot]
         lqt = Mock(lot_quantity=Decimal('1'))
+        matched_lqt = Mock(lot_quantity=Decimal('9'))
 
         lot_model = Mock()
         lotqt_model = Mock()
-        lotqt_model.search.return_value = [lqt]
+        lotqt_model.search.side_effect = [[lqt], [matched_lqt]]
 
         with patch(
                 'trytond.modules.purchase_trade.sale.Pool'
@@ -1123,6 +1124,48 @@ class PurchaseTradeTestCase(ModuleTestCase):
             with self.assertRaises(UserError):
                 SaleLine.write([line], {'quantity_theorical': Decimal('8')})
 
+    def test_purchase_line_write_syncs_open_lot_qt_with_physical_lots(self):
+        'purchase line write keeps open lot.qt net of existing physical lots'
+        PurchaseLine = Pool().get('purchase.line')
+        line = Mock(id=4, quantity_theorical=Decimal('10000'))
+        line.unit = Mock()
+        vlot = Mock(id=102, lot_type='virtual')
+        vlot.get_current_quantity_converted.return_value = Decimal('10000')
+        physical = Mock(lot_type='physic')
+        physical.get_current_quantity_converted.return_value = Decimal('10000')
+        line.lots = [vlot, physical]
+        lqt = Mock(lot_quantity=Decimal('10000'))
+
+        lot_model = Mock()
+        lotqt_model = Mock()
+        lotqt_model.search.side_effect = [[lqt], []]
+
+        with patch(
+                'trytond.modules.purchase_trade.purchase.Pool'
+                ) as PoolMock, patch(
+                    'trytond.modules.purchase_trade.purchase.super'
+                ) as super_mock:
+            PoolMock.return_value.get.side_effect = lambda name: {
+                'lot.lot': lot_model,
+                'lot.qt': lotqt_model,
+            }[name]
+
+            def fake_super_write(*args):
+                for records, values in zip(args[::2], args[1::2]):
+                    if 'quantity_theorical' in values:
+                        for record in records:
+                            record.quantity_theorical = values['quantity_theorical']
+
+            super_mock.return_value.write.side_effect = fake_super_write
+
+            PurchaseLine.write(
+                [line], {'quantity_theorical': Decimal('20000')})
+
+        self.assertEqual(lqt.lot_quantity, Decimal('10000'))
+        vlot.set_current_quantity.assert_not_called()
+        lot_model.save.assert_not_called()
+        lotqt_model.save.assert_not_called()
+
     def test_purchase_line_write_initial_theorical_qty_does_not_double_open_lot(self):
         'purchase line write does not re-add quantity when initializing contractual qty'
         PurchaseLine = Pool().get('purchase.line')

notes/business_rules.md

@@ -165,3 +165,18 @@ elle existe, par exemple:
 - La ligne `Estimated date` avec `trigger = bldate` reste le support metier
   pour porter `fin_int_delta`; la date estimee elle-meme ne sert pas au calcul
   du montant `% rate`.
+
+## Session 2026-05-01 - Solde ouvert apres lots physiques
+
+### `purchase.line` / `sale.line` et `lot.qt`
+
+- Une modification de `quantity_theorical` ne doit plus ajuster le `lot.qt`
+  libre uniquement par delta.
+- Le solde ouvert est recalcule en tenant compte des lots physiques deja crees:
+  `quantity_theorical - somme(lots physiques)`.
+- Le `lot.qt` libre non matche / non shippe est ensuite aligne sur ce solde
+  moins les `lot.qt` deja matches ou shippes.
+- Si ce calcul donne un solde negatif, la modification est bloquee avec
+  `Please unlink or unmatch lot`.
+- Cas de reference: une ligne achat passee de `10000` a `20000` avec deja
+  `10000` physiques doit rester a `10000` ouverts et `10000` physiques.