Remove physical lot for Planned

This commit is contained in:
2026-06-13 10:38:19 +02:00
parent 1f3e32fbe8
commit 77b1ab333b
3 changed files with 352 additions and 16 deletions

View File

@@ -254,9 +254,22 @@ class Lot(metaclass=PoolMeta):
delta_qt = fields.Numeric("Delta Qt")
delta_pr = fields.Numeric("Delta Pr")
delta_amt = fields.Numeric("Delta Amt")
warrant_nb = fields.Char("Warrant Nb")
lot_chunk_key = fields.Integer("Chunk key")
#fees = fields.Many2Many('fee.lots', 'lot', 'fee',"Fees")
warrant_nb = fields.Char("Warrant Nb")
lot_chunk_key = fields.Integer("Chunk key")
planned_from_location = fields.Many2One(
'stock.location', "Planned From")
planned_to_location = fields.Many2One(
'stock.location', "Planned To")
planned_transport_type = fields.Selection([
(None, ''),
('vessel', "Vessel"),
('truck', "Truck"),
('container', "Container"),
], "Planned Transport")
planned_from_date = fields.Date("Planned From Date")
planned_to_date = fields.Date("Planned To Date")
planned_note = fields.Char("Planning Note")
#fees = fields.Many2Many('fee.lots', 'lot', 'fee',"Fees")
dashboard = fields.Many2One('purchase.dashboard',"Dashboard")
pivot = fields.Function(
fields.Text('Pivot'), 'get_pivot'
@@ -265,8 +278,17 @@ class Lot(metaclass=PoolMeta):
fields.Text('Graphe JSON'), 'get_vis_data'
)
flow_graph = fields.Function(fields.Text("Flux comptable (graph)"),
'get_flow_graph')
flow_graph = fields.Function(fields.Text("Flux comptable (graph)"),
'get_flow_graph')
def is_planned_transport(self):
return bool(
self.planned_from_location
or self.planned_to_location
or self.planned_transport_type
or self.planned_from_date
or self.planned_to_date
or self.planned_note)
def get_pivot(self,name=None):
AccountMoveLine = Pool().get('account.move.line')
@@ -1352,6 +1374,151 @@ class LotQt(
cls.create([cls._cancel_plan_values(planned)])
cls.delete([planned])
@classmethod
def _planned_open_domain(cls, planned):
return [
('lot_p', '=', planned.lot_p.id if planned.lot_p else None),
('lot_s', '=', planned.lot_s.id if planned.lot_s else None),
('lot_shipment_in', '=', None),
('lot_shipment_out', '=', None),
('lot_shipment_internal', '=', None),
('planned_from_location', '=',
planned.planned_from_location.id
if planned.planned_from_location else None),
('planned_to_location', '=',
planned.planned_to_location.id
if planned.planned_to_location else None),
('planned_transport_type', '=',
planned.planned_transport_type),
('planned_from_date', '=', planned.planned_from_date),
('planned_to_date', '=', planned.planned_to_date),
('planned_note', '=', planned.planned_note),
]
@classmethod
def _planned_open_values(cls, planned, quantity):
return {
'lot_p': planned.lot_p.id if planned.lot_p else None,
'lot_s': planned.lot_s.id if planned.lot_s else None,
'lot_quantity': quantity,
'lot_unit': planned.lot_unit.id if planned.lot_unit else None,
'lot_av': planned.lot_av,
'lot_status': planned.lot_status,
'lot_visible': planned.lot_visible,
'planned_from_location': (
planned.planned_from_location.id
if planned.planned_from_location else None),
'planned_to_location': (
planned.planned_to_location.id
if planned.planned_to_location else None),
'planned_transport_type': planned.planned_transport_type,
'planned_from_date': planned.planned_from_date,
'planned_to_date': planned.planned_to_date,
'planned_note': planned.planned_note,
}
@classmethod
def restore_planned_open_quantity(cls, planned, quantity):
targets = cls.search(cls._planned_open_domain(planned), limit=1)
if targets:
target = targets[0]
target.lot_quantity = (
Decimal(str(target.lot_quantity or 0))
+ Decimal(str(quantity or 0)))
cls.save([target])
else:
cls.create([cls._planned_open_values(planned, quantity)])
@classmethod
def _shipment_domain_from_origin(cls, shipment_origin):
values = {
'lot_shipment_in': None,
'lot_shipment_out': None,
'lot_shipment_internal': None,
}
if not shipment_origin:
return values
str_shipment = str(shipment_origin)
index = str_shipment.rfind(",")
if index == -1:
return values
shipment_id = int(str_shipment[index + 1:].strip())
if 'stock.shipment.in' in str_shipment:
values['lot_shipment_in'] = shipment_id
elif 'stock.shipment.out' in str_shipment:
values['lot_shipment_out'] = shipment_id
elif 'stock.shipment.internal' in str_shipment:
values['lot_shipment_internal'] = shipment_id
return values
@classmethod
def _planned_scheduled_domain(cls, physical_lot, quantity_lot_s,
shipment_origin):
shipment_values = cls._shipment_domain_from_origin(shipment_origin)
return [
('lot_p', '=', physical_lot.getVlot_p().id),
('lot_s', '=',
quantity_lot_s.id if quantity_lot_s else None),
('lot_shipment_in', '=',
shipment_values['lot_shipment_in']),
('lot_shipment_out', '=',
shipment_values['lot_shipment_out']),
('lot_shipment_internal', '=',
shipment_values['lot_shipment_internal']),
('planned_from_location', '=',
physical_lot.planned_from_location.id
if physical_lot.planned_from_location else None),
('planned_to_location', '=',
physical_lot.planned_to_location.id
if physical_lot.planned_to_location else None),
('planned_transport_type', '=',
physical_lot.planned_transport_type),
('planned_from_date', '=', physical_lot.planned_from_date),
('planned_to_date', '=', physical_lot.planned_to_date),
('planned_note', '=', physical_lot.planned_note),
]
@classmethod
def _planned_scheduled_values(cls, physical_lot, quantity, quantity_lot_s,
shipment_origin):
values = {
'lot_p': physical_lot.getVlot_p().id,
'lot_s': quantity_lot_s.id if quantity_lot_s else None,
'lot_quantity': quantity,
'lot_unit': (
physical_lot.lot_unit_line.id
if physical_lot.lot_unit_line else None),
'lot_av': 'available',
'lot_status': 'forecast',
'planned_from_location': (
physical_lot.planned_from_location.id
if physical_lot.planned_from_location else None),
'planned_to_location': (
physical_lot.planned_to_location.id
if physical_lot.planned_to_location else None),
'planned_transport_type': physical_lot.planned_transport_type,
'planned_from_date': physical_lot.planned_from_date,
'planned_to_date': physical_lot.planned_to_date,
'planned_note': physical_lot.planned_note,
}
values.update(cls._shipment_domain_from_origin(shipment_origin))
return values
@classmethod
def restore_planned_scheduled_quantity(cls, physical_lot, quantity,
quantity_lot_s, shipment_origin):
targets = cls.search(cls._planned_scheduled_domain(
physical_lot, quantity_lot_s, shipment_origin), limit=1)
if targets:
target = targets[0]
target.lot_quantity = (
Decimal(str(target.lot_quantity or 0))
+ Decimal(str(quantity or 0)))
cls.save([target])
else:
cls.create([cls._planned_scheduled_values(
physical_lot, quantity, quantity_lot_s, shipment_origin)])
def isVirtualP(self):
if self.lot_p:
Lot = Pool().get('lot.lot')
@@ -1551,12 +1718,16 @@ class LotQt(
break
qt_p = left_qt
@classmethod
@classmethod
def add_physical_lots(cls,lqt,vlots):
Purchase = Pool().get('purchase.purchase')
LotQt = Pool().get('lot.qt')
Lot = Pool().get('lot.lot')
Uom = Pool().get('product.uom')
if not lqt.has_shipment():
raise UserError(
"Please link the quantity to transport before adding "
"physical lots.")
purchase = lqt.lot_p.line.purchase
lots = []
tot_qt = 0
@@ -1795,8 +1966,14 @@ class LotQt(
newlot.lot_unit = l.lot_unit
newlot.lot_unit_line = l.lot_unit_line
newlot.lot_premium = getattr(l, 'lot_premium', None) or Decimal(0)
if newlot.lot_shipment_in or newlot.lot_shipment_internal or newlot.lot_shipment_out:
newlot.lot_status = 'transit'
newlot.planned_from_location = self.planned_from_location
newlot.planned_to_location = self.planned_to_location
newlot.planned_transport_type = self.planned_transport_type
newlot.planned_from_date = self.planned_from_date
newlot.planned_to_date = self.planned_to_date
newlot.planned_note = self.planned_note
if newlot.lot_shipment_in or newlot.lot_shipment_internal or newlot.lot_shipment_out:
newlot.lot_status = 'transit'
newlot.set_current_quantity(l.lot_quantity,l.lot_gross_quantity,1)
if (newlot.lot_shipment_in or newlot.lot_shipment_internal or newlot.lot_shipment_out):
newlot.set_current_quantity(l.lot_quantity,l.lot_gross_quantity,2)
@@ -3862,7 +4039,9 @@ class LotUnship(Wizard):
#Decrease forecasted virtual part shipped
lot.updateVirtualPart(-qt,lqt.lot_shipment_origin,lqt.lot_s,mode)
#Increase forecasted virtual part non shipped
if not lot.updateVirtualPart(qt,None,lqt.lot_s,mode):
if lqt.is_planned_transport():
LotQt.restore_planned_open_quantity(lqt, qt)
elif not lot.updateVirtualPart(qt,None,lqt.lot_s,mode):
lot.createVirtualPart(qt,None,lqt.lot_s,mode)
if lqt.lot_p and lqt.lot_p.line:
affected_lines.append(lqt.lot_p.line)
@@ -3928,12 +4107,16 @@ class LotAdding(Wizard):
ids = context.get('active_ids')
for i in ids:
val = {}
if i > 10000000 :
lqt = LotQt(i - 10000000)
if not lqt.lot_p:
raise UserError("To add physical lot to sale, use action 'Apply matching'!")
lot = lqt.lot_p
unit = lot.lot_unit_line.id
if i > 10000000 :
lqt = LotQt(i - 10000000)
if not lqt.lot_p:
raise UserError("To add physical lot to sale, use action 'Apply matching'!")
if not lqt.has_shipment():
raise UserError(
"Please link the quantity to transport before adding "
"physical lots.")
lot = lqt.lot_p
unit = lot.lot_unit_line.id
quantity = lqt.lot_quantity
sh_in = lqt.lot_shipment_in.id if lqt.lot_shipment_in else None
sh_int = lqt.lot_shipment_internal.id if lqt.lot_shipment_internal else None
@@ -4127,6 +4310,7 @@ class LotRemove(Wizard):
def transition_start(self):
Lot = Pool().get('lot.lot')
LotQt = Pool().get('lot.qt')
Move = Pool().get('stock.move')
Warning = Pool().get('res.user.warning')
for r in self.records:
@@ -4148,7 +4332,10 @@ class LotRemove(Wizard):
#delete move first if exist
if lot.move:
Move.delete([lot.move])
if not lot.updateVirtualPart(quantity, shipment_origin, lot_s):
if lot.is_planned_transport():
LotQt.restore_planned_scheduled_quantity(
lot, quantity, lot_s, shipment_origin)
elif not lot.updateVirtualPart(quantity, shipment_origin, lot_s):
lot.createVirtualPart(quantity, shipment_origin, lot_s)
Lot.delete([lot])
return 'end'

View File

@@ -494,6 +494,143 @@ class PurchaseTradeTestCase(ModuleTestCase):
Lot.assert_lines_quantity_consistency.assert_called_once_with([
lot.line])
def test_lot_unship_restores_planned_open_quantity(self):
'unlinking a planned scheduled lot.qt returns it to planned'
wizard = lot_module.LotUnship()
record = Mock(
id=10000007,
r_lot_type='virtual',
)
wizard.records = [record]
purchase_line = Mock()
sale_line = Mock()
lot = Mock(line=purchase_line)
sale_lot = Mock(sale_line=sale_line)
lqt = Mock(
lot_quantity=Decimal('40'),
lot_p=lot,
lot_s=sale_lot,
lot_shipment_origin='stock.shipment.in,12',
)
lqt.is_planned_transport.return_value = True
Lot = Mock()
Lot.return_value = Mock()
Lot.skip_quantity_consistency.return_value.__enter__ = Mock()
Lot.skip_quantity_consistency.return_value.__exit__ = Mock()
LotQt = Mock()
LotQt.return_value = lqt
Move = Mock()
pool = Mock()
pool.get.side_effect = [Lot, LotQt, Move]
with patch('trytond.modules.purchase_trade.lot.Pool',
return_value=pool):
state = wizard.transition_start()
self.assertEqual(state, 'end')
lot.updateVirtualPart.assert_called_once_with(
Decimal('-40'), 'stock.shipment.in,12', sale_lot, 'both')
LotQt.restore_planned_open_quantity.assert_called_once_with(
lqt, Decimal('40'))
lot.createVirtualPart.assert_not_called()
Lot.assert_lines_quantity_consistency.assert_called_once_with([
purchase_line, sale_line])
def test_add_physical_lots_requires_linked_transport(self):
'physical lots can only be added from a scheduled quantity'
LotQt = Pool().get('lot.qt')
lqt = Mock()
lqt.has_shipment.return_value = False
with self.assertRaises(UserError):
LotQt.add_physical_lots(lqt, [])
def test_add_physical_lot_copies_planned_transport_memory(self):
'physical lots keep planned transport fields from lot.qt'
LotQt = Pool().get('lot.qt')
lqt = LotQt()
planned_from = Mock()
planned_to = Mock()
purchase_line = Mock(product=Mock(), fees=[])
lqt.lot_p = Mock(line=purchase_line)
lqt.lot_s = None
lqt.lot_shipment_in = Mock(fees=[])
lqt.lot_shipment_internal = None
lqt.lot_shipment_out = None
lqt.planned_from_location = planned_from
lqt.planned_to_location = planned_to
lqt.planned_transport_type = 'truck'
lqt.planned_from_date = datetime.date(2026, 6, 15)
lqt.planned_to_date = datetime.date(2026, 6, 20)
lqt.planned_note = 'Gate 4'
physical_input = Mock(
lot_qt=Decimal('1'),
lot_unit=Mock(),
lot_unit_line=Mock(),
lot_quantity=Decimal('40'),
lot_gross_quantity=Decimal('41'),
lot_premium=Decimal('0'),
lot_chunk_key=None,
)
newlot = Mock()
newlot.line = purchase_line
newlot.sale_line = None
newlot.set_current_quantity = Mock()
newlot.get_current_quantity_converted.return_value = Decimal('40')
Lot = Mock()
Lot.return_value = newlot
FeeLots = Mock()
Purchase = Mock()
pool = Mock()
pool.get.side_effect = [Lot, FeeLots, Purchase]
with patch('trytond.modules.purchase_trade.lot.Pool',
return_value=pool):
lot, _ = lqt.add_physical_lot(physical_input)
self.assertIs(lot.planned_from_location, planned_from)
self.assertIs(lot.planned_to_location, planned_to)
self.assertEqual(lot.planned_transport_type, 'truck')
self.assertEqual(lot.planned_from_date, datetime.date(2026, 6, 15))
self.assertEqual(lot.planned_to_date, datetime.date(2026, 6, 20))
self.assertEqual(lot.planned_note, 'Gate 4')
def test_remove_physical_lot_restores_planned_scheduled_quantity(self):
'removing a planned physical lot restores the planned scheduled lot.qt'
wizard = lot_module.LotRemove()
wizard.records = [Mock(id=2014)]
quantity = Decimal('40')
lot_s = Mock()
lot = Mock(
move=None,
lot_shipment_origin='stock.shipment.in,12',
lot_shipment_in=None,
lot_shipment_internal=None,
lot_shipment_out=None,
)
lot.get_current_quantity_converted.return_value = quantity
lot.getVlot_s.return_value = lot_s
lot.sale_line = Mock()
lot.is_planned_transport.return_value = True
Lot = Mock()
Lot.return_value = lot
LotQt = Mock()
Move = Mock()
Warning = Mock()
Warning.check.return_value = False
pool = Mock()
pool.get.side_effect = [Lot, LotQt, Move, Warning]
with patch('trytond.modules.purchase_trade.lot.Pool',
return_value=pool):
state = wizard.transition_start()
self.assertEqual(state, 'end')
LotQt.restore_planned_scheduled_quantity.assert_called_once_with(
lot, quantity, lot_s, 'stock.shipment.in,12')
lot.updateVirtualPart.assert_not_called()
Lot.delete.assert_called_once_with([lot])
def test_lot_shipping_scheduled_lotqt_keeps_planned_locations(self):
'scheduled lot.qt keeps the planned from and to locations'
wizard = lot_module.LotShipping()

View File

@@ -1,6 +1,18 @@
<?xml version="1.0"?>
<data>
<xpath expr="/form/label[@name='warrant_nb']" position="before">
<label name="planned_from_location"/>
<field name="planned_from_location"/>
<label name="planned_to_location"/>
<field name="planned_to_location"/>
<label name="planned_transport_type"/>
<field name="planned_transport_type"/>
<label name="planned_from_date"/>
<field name="planned_from_date"/>
<label name="planned_to_date"/>
<field name="planned_to_date"/>
<label name="planned_note"/>
<field name="planned_note"/>
<label name="sale_invoice_padding"/>
<field name="sale_invoice_padding"/>
</xpath>