admin/tradon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add physical + Link

Browse Source
This commit is contained in:
laurentbarontini
2026-06-22 15:33:43 +02:00
parent 8f676c2396
commit a880535bc4
4 changed files with 391 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

255
modules/purchase_trade/lot.py
View File

@@ -1210,8 +1210,8 @@ class Lot(metaclass=PoolMeta):
# super().write(*args)
# lots = sum(args[::2], [])
class LotQt(
ModelSQL, ModelView):
class LotQt(
ModelSQL, ModelView):
"Lots Qt"
__name__ = 'lot.qt'
@@ -1356,6 +1356,72 @@ class LotQt(
or self.lot_shipment_out
or self.lot_shipment_internal)
@classmethod
def link_to_transport(cls, source, quantity, shipment_origin):
quantity = abs(Decimal(str(quantity or 0))).quantize(
Decimal("0.00001"))
available = abs(Decimal(str(source.lot_quantity or 0))).quantize(
Decimal("0.00001"))
if not quantity or quantity > available:
raise UserError("Quantity to link exceeds available quantity.")
shipment_model, shipment_id = str(shipment_origin).split(',', 1)
shipment_field = {
'stock.shipment.in': 'lot_shipment_in',
'stock.shipment.out': 'lot_shipment_out',
'stock.shipment.internal': 'lot_shipment_internal',
}.get(shipment_model)
if not shipment_field:
raise UserError("Shipment type is not supported.")
shipment_id = int(shipment_id)
signed_quantity = quantity
if Decimal(str(source.lot_quantity or 0)) < 0:
signed_quantity = -quantity
if source.is_planned_transport():
values = cls._split_plan_values(
source, signed_quantity, {
'planned_from_location': cls._record_id(
source.planned_from_location),
'planned_to_location': cls._record_id(
source.planned_to_location),
'planned_transport_type': source.planned_transport_type,
'planned_from_date': source.planned_from_date,
'planned_to_date': source.planned_to_date,
'planned_note': source.planned_note,
})
values[shipment_field] = shipment_id
linked, = cls.create([values])
source.lot_quantity -= signed_quantity
if abs(source.lot_quantity) < Decimal("0.00001"):
source.lot_quantity = 0
cls.save([source])
return linked
plan_lot = source.lot_p or source.lot_s
plan_mode = 'both' if source.lot_p else 'only sale'
if not plan_lot.updateVirtualPart(
signed_quantity, shipment_origin,
source.lot_s, plan_mode):
plan_lot.createVirtualPart(
signed_quantity, shipment_origin,
source.lot_s, plan_mode)
plan_lot.updateVirtualPart(
-signed_quantity, None, source.lot_s, plan_mode)
domain = [
('lot_p', '=', source.lot_p),
('lot_s', '=', source.lot_s),
(shipment_field, '=', shipment_id),
]
linked = cls.search(domain)
if not linked:
raise UserError("The linked transport quantity was not created.")
return linked[0]
@staticmethod
def _record_id(record):
return record.id if record else None
@classmethod
def _split_plan_values(cls, source, quantity, values):
return {
@@ -3302,7 +3368,7 @@ class LotContext(ModelView):
return super().get_context(fields_names=fields_names)
class LotShipping(Wizard):
class LotShipping(Wizard):
"Link lot to transport"
__name__ = "lot.shipping"
@@ -3325,14 +3391,20 @@ class LotShipping(Wizard):
def default_ship(self, fields):
values = {}
planned_from_locations = {
record.r_planned_from_location.id
location.id
for record in self.records
if record.r_planned_from_location
for location in [
record.r_planned_from_location
or getattr(record.r_purchase, 'from_location', None)]
if location
}
planned_to_locations = {
record.r_planned_to_location.id
location.id
for record in self.records
if record.r_planned_to_location
for location in [
record.r_planned_to_location
or getattr(record.r_purchase, 'to_location', None)]
if location
}
if len(planned_from_locations) == 1:
values['planned_from_location'] = next(iter(
@@ -3368,6 +3440,12 @@ class LotShipping(Wizard):
values['from_location'] = self.ship.planned_from_location.id
if self.ship.planned_to_location:
values['to_location'] = self.ship.planned_to_location.id
bl_number = getattr(self.ship, 'shipment_bl_number', None)
bl_date = getattr(self.ship, 'shipment_bl_date', None)
if bl_number:
values['bl_number'] = bl_number
if bl_date:
values['bl_date'] = bl_date
return values
def _create_shipment_in(self):
@@ -3375,34 +3453,6 @@ class LotShipping(Wizard):
shipment, = ShipmentIn.create([self._shipment_in_create_values()])
self.ship.shipment_in = shipment
def _scheduled_lotqt_values(self, lqt, quantity):
values = {
'lot_p': self._record_id(lqt.lot_p),
'lot_s': self._record_id(lqt.lot_s),
'lot_quantity': quantity,
'lot_unit': self._record_id(lqt.lot_unit),
'lot_av': lqt.lot_av,
'lot_status': lqt.lot_status,
'lot_visible': lqt.lot_visible,
'planned_from_location': self._record_id(
lqt.planned_from_location),
'planned_to_location': self._record_id(lqt.planned_to_location),
'planned_transport_type': lqt.planned_transport_type,
'planned_from_date': lqt.planned_from_date,
'planned_to_date': lqt.planned_to_date,
'planned_note': lqt.planned_note,
}
if self.ship.shipment == 'in':
values['lot_shipment_in'] = self.ship.shipment_in.id
elif self.ship.shipment == 'out':
values['lot_shipment_out'] = self.ship.shipment_out.id
elif self.ship.shipment == 'int':
values['lot_shipment_internal'] = self.ship.shipment_internal.id
return values
def _create_scheduled_lotqt(self, LotQt, lqt, quantity):
LotQt.create([self._scheduled_lotqt_values(lqt, quantity)])
def transition_shipping(self):
Lot = Pool().get('lot.lot')
LotQt = Pool().get('lot.qt')
@@ -3478,41 +3528,8 @@ class LotShipping(Wizard):
affected_lines.append(l.sale_line)
else:
lqt = LotQt(r.id - 10000000)
available_to_ship = abs(Decimal(str(
lqt.lot_quantity or 0)))
if shipped_quantity > available_to_ship:
raise UserError(
"Quantity to link exceeds available quantity.")
plan_lot = lqt.lot_p
plan_mode = 'both'
if not plan_lot:
plan_lot = lqt.lot_s
plan_mode = 'only sale'
signed_shipped_quantity = shipped_quantity
if Decimal(str(lqt.lot_quantity or 0)) < 0:
signed_shipped_quantity = -shipped_quantity
#Increase forecasted virtual part shipped
if lqt.is_planned_transport():
self._create_scheduled_lotqt(
LotQt, lqt, signed_shipped_quantity)
else:
if not plan_lot.updateVirtualPart(
signed_shipped_quantity, shipment_origin,
lqt.lot_s, plan_mode):
logger.info("LotShipping2:%s", shipped_quantity)
plan_lot.createVirtualPart(
signed_shipped_quantity, shipment_origin,
lqt.lot_s, plan_mode)
#Decrease forecasted virtual part non shipped
if lqt.is_planned_transport():
lqt.lot_quantity -= signed_shipped_quantity
if abs(lqt.lot_quantity) < Decimal("0.00001"):
lqt.lot_quantity = 0
LotQt.save([lqt])
else:
plan_lot.updateVirtualPart(
-signed_shipped_quantity, None,
lqt.lot_s, plan_mode)
LotQt.link_to_transport(
lqt, shipped_quantity, shipment_origin)
if lqt.lot_p and lqt.lot_p.line:
affected_lines.append(lqt.lot_p.line)
if lqt.lot_s and lqt.lot_s.sale_line:
@@ -4261,7 +4278,7 @@ class LotUnship(Wizard):
def end(self):
return 'reload'
class LotAdding(Wizard):
class LotAdding(Wizard):
"Adding physical lots"
__name__ = "lot.add"
@@ -4295,14 +4312,11 @@ class LotAdding(Wizard):
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
purchase = lot.line.purchase
unit = lot.lot_unit_line.id
quantity = lqt.lot_quantity
sh_in = lqt.lot_shipment_in.id if lqt.lot_shipment_in else None
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
sh_out = lqt.lot_shipment_out.id if lqt.lot_shipment_out else None
else:
@@ -4314,12 +4328,58 @@ class LotAdding(Wizard):
'shipment_in':sh_in,
'shipment_internal':sh_int,
'shipment_out':sh_out,
'transport_missing': not any((sh_in, sh_int, sh_out)),
'shipment_supplier': (
purchase.party.id
if not any((sh_in, sh_int, sh_out)) and purchase.party
else None),
'transport_from_location': (
(lqt.planned_from_location or purchase.from_location).id
if not any((sh_in, sh_int, sh_out))
and (lqt.planned_from_location or purchase.from_location)
else None),
'transport_to_location': (
(lqt.planned_to_location or purchase.to_location).id
if not any((sh_in, sh_int, sh_out))
and (lqt.planned_to_location or purchase.to_location)
else None),
}
def transition_adding(self):
def transition_adding(self):
if self.records:
LotQt = Pool().get('lot.qt')
lqt = LotQt(self.records[0].id - 10000000)
if not lqt.has_shipment():
shipment = self.add.shipment_in
if self.add.create_new_shipment and shipment:
raise UserError(
"Choose either an existing shipment or create a new "
"shipment.")
if self.add.create_new_shipment:
if not self.add.shipment_supplier:
raise UserError("Shipment supplier is required.")
ShipmentIn = Pool().get('stock.shipment.in')
values = {
'supplier': self.add.shipment_supplier.id,
}
if self.add.transport_from_location:
values['from_location'] = (
self.add.transport_from_location.id)
if self.add.transport_to_location:
values['to_location'] = self.add.transport_to_location.id
if self.add.shipment_bl_number:
values['bl_number'] = self.add.shipment_bl_number
if self.add.shipment_bl_date:
values['bl_date'] = self.add.shipment_bl_date
shipment, = ShipmentIn.create([values])
if not shipment:
raise UserError(
"Select a Shipment In or create a new shipment.")
Lot = Pool().get('lot.lot')
with Lot.skip_quantity_consistency():
lqt = LotQt.link_to_transport(
lqt, abs(Decimal(str(lqt.lot_quantity or 0))),
'stock.shipment.in,%s' % shipment.id)
LotQt.add_physical_lots(
lqt, self.add.lots, self.add.unlink_remainder)
return 'end'
@@ -4327,13 +4387,44 @@ class LotAdding(Wizard):
def end(self):
return 'reload'
class LotAddLot(ModelView):
class LotAddLot(ModelView):
"Add physical lots"
__name__ = "lot.add.lot"
unit = fields.Many2One('product.uom',"Unit",readonly=True)
quantity = fields.Numeric("Quantity available",digits='unit',readonly=True)
unlink_remainder = fields.Boolean("Unlink remaining quantity")
shipment_in = fields.Many2One('stock.shipment.in',"Shipment In",states={'invisible': ~Eval('shipment_in')})
transport_missing = fields.Boolean("Transport missing", readonly=True)
create_new_shipment = fields.Boolean(
"Create a new shipment",
states={'invisible': ~Eval('transport_missing')})
shipment_supplier = fields.Many2One(
'party.party', "Shipment supplier",
states={
'invisible': ~(
Eval('transport_missing') & Eval('create_new_shipment')),
'required': (
Eval('transport_missing') & Eval('create_new_shipment')),
})
transport_from_location = fields.Many2One(
'stock.location', "From location", readonly=True,
states={'invisible': ~Eval('transport_missing')})
transport_to_location = fields.Many2One(
'stock.location', "To location", readonly=True,
states={'invisible': ~Eval('transport_missing')})
shipment_bl_number = fields.Char(
"BL number",
states={'invisible': ~(
Eval('transport_missing') & Eval('create_new_shipment'))})
shipment_bl_date = fields.Date(
"BL date",
states={'invisible': ~(
Eval('transport_missing') & Eval('create_new_shipment'))})
shipment_in = fields.Many2One(
'stock.shipment.in', "Shipment In",
states={
'invisible': ~Eval('transport_missing') & ~Eval('shipment_in'),
'readonly': ~Eval('transport_missing'),
})
shipment_internal = fields.Many2One('stock.shipment.internal',"Shipment Internal",states={'invisible': ~Eval('shipment_internal')})
shipment_out = fields.Many2One('stock.shipment.out',"Shipment Out",states={'invisible': ~Eval('shipment_out')})
lots = fields.One2Many('lot.add.line','lal',"Lots")
@@ -4341,6 +4432,10 @@ class LotAddLot(ModelView):
@classmethod
def default_unlink_remainder(cls):
return True
@classmethod
def default_create_new_shipment(cls):
return False
class LotAddLine(ModelView):
"Lots"

8
modules/purchase_trade/stock.py
View File

@@ -713,8 +713,12 @@ class CharterParty(ModelSQL, ModelView):
class ShipmentIn(metaclass=PoolMeta):
__name__ = 'stock.shipment.in'
from_location = fields.Many2One('stock.location', 'From location')
to_location = fields.Many2One('stock.location', 'To location')
from_location = fields.Many2One(
'stock.location', 'From location',
domain=[('type', '!=', 'customer')])
to_location = fields.Many2One(
'stock.location', 'To location',
domain=[('type', '!=', 'supplier')])
shipment_type = fields.Function(fields.Selection([
(None, ''),
('dropship', 'Dropship'),

207
modules/purchase_trade/tests/test_module.py
View File

@@ -372,6 +372,22 @@ class PurchaseTradeTestCase(ModuleTestCase):
'planned_to_location': 20,
})
def test_lot_shipping_defaults_purchase_locations_without_plan(self):
'link to transport falls back to purchase locations without a plan'
wizard = lot_module.LotShipping()
wizard.records = [Mock(
r_planned_from_location=None,
r_planned_to_location=None,
r_purchase=Mock(
from_location=Mock(id=10),
to_location=Mock(id=20)),
)]
self.assertEqual(wizard.default_ship(None), {
'planned_from_location': 10,
'planned_to_location': 20,
})
def test_lot_shipping_defaults_uniform_planned_locations(self):
'link to transport carries uniform planned locations from selection'
wizard = lot_module.LotShipping()
@@ -509,6 +525,47 @@ class PurchaseTradeTestCase(ModuleTestCase):
Lot.assert_lines_quantity_consistency.assert_called_once_with([
lot.line])
def test_lot_shipping_open_quantity_uses_shared_transport_link(self):
'standalone link action keeps linking open lot.qt quantities'
wizard = lot_module.LotShipping()
shipment = Mock(id=40)
wizard.ship = Mock(
shipment='in',
shipment_in=shipment,
shipment_out=None,
shipment_internal=None,
create_new_shipment=False,
quantity=None,
)
wizard.records = [Mock(
id=10000007,
r_lot_shipment_in=None,
r_lot_quantity=Decimal('125'),
r_lot_matched=Decimal('0'),
)]
purchase_line = Mock()
source = Mock(
lot_p=Mock(line=purchase_line),
lot_s=None,
)
Lot = Mock()
Lot.skip_quantity_consistency.return_value.__enter__ = Mock()
Lot.skip_quantity_consistency.return_value.__exit__ = Mock()
LotQt = Mock(return_value=source)
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_shipping()
self.assertEqual(state, 'end')
LotQt.link_to_transport.assert_called_once_with(
source, Decimal('125.00000'), 'stock.shipment.in,40')
Lot.assert_lines_quantity_consistency.assert_called_once_with([
purchase_line])
def test_lot_unship_restores_planned_open_quantity(self):
'unlinking a planned scheduled lot.qt returns it to planned'
wizard = lot_module.LotUnship()
@@ -648,14 +705,10 @@ class PurchaseTradeTestCase(ModuleTestCase):
def test_lot_shipping_scheduled_lotqt_keeps_planned_locations(self):
'scheduled lot.qt keeps the planned from and to locations'
wizard = lot_module.LotShipping()
wizard.ship = Mock(
shipment='in',
shipment_in=Mock(id=40),
shipment_out=None,
shipment_internal=None,
)
LotQt = Pool().get('lot.qt')
linked = Mock()
lqt = Mock(
lot_quantity=Decimal('125'),
lot_p=Mock(id=1),
lot_s=Mock(id=2),
lot_unit=Mock(id=3),
@@ -669,11 +722,15 @@ class PurchaseTradeTestCase(ModuleTestCase):
planned_to_date=datetime.date(2026, 7, 5),
planned_note='POL to POD',
)
LotQt = Mock()
lqt.is_planned_transport.return_value = True
wizard._create_scheduled_lotqt(LotQt, lqt, Decimal('125'))
with patch.object(
LotQt, 'create', return_value=[linked]) as create, \
patch.object(LotQt, 'save') as save:
result = LotQt.link_to_transport(
lqt, Decimal('125'), 'stock.shipment.in,40')
LotQt.create.assert_called_once_with([{
create.assert_called_once_with([{
'lot_p': 1,
'lot_s': 2,
'lot_quantity': Decimal('125'),
@@ -689,6 +746,124 @@ class PurchaseTradeTestCase(ModuleTestCase):
'planned_note': 'POL to POD',
'lot_shipment_in': 40,
}])
save.assert_called_once_with([lqt])
self.assertEqual(lqt.lot_quantity, 0)
self.assertIs(result, linked)
def test_add_physical_lot_defaults_transport_from_purchase(self):
'add physical lot proposes shipment creation from purchase values'
wizard = lot_module.LotAdding()
purchase = Mock(
party=Mock(id=30),
from_location=Mock(id=10),
to_location=Mock(id=20),
)
lqt = Mock(
lot_p=Mock(line=Mock(
purchase=purchase,
unit=Mock(id=3))),
lot_quantity=Decimal('125'),
lot_shipment_in=None,
lot_shipment_internal=None,
lot_shipment_out=None,
planned_from_location=None,
planned_to_location=None,
)
lqt.lot_p.lot_unit_line = Mock(id=3)
LotQt = Mock(return_value=lqt)
pool = Mock()
pool.get.side_effect = [Mock(), LotQt]
with patch('trytond.modules.purchase_trade.lot.Pool',
return_value=pool), patch(
'trytond.modules.purchase_trade.lot.Transaction'
) as TransactionMock:
TransactionMock.return_value.context = {
'active_ids': [10000007]}
values = wizard.default_add(None)
self.assertTrue(values['transport_missing'])
self.assertEqual(values['shipment_supplier'], 30)
self.assertEqual(values['transport_from_location'], 10)
self.assertEqual(values['transport_to_location'], 20)
def test_add_physical_lot_links_existing_shipment_before_add(self):
'add physical lot can link an existing shipment in one transaction'
wizard = lot_module.LotAdding()
shipment = Mock(id=40)
source = Mock(lot_quantity=Decimal('125'))
source.has_shipment.return_value = False
linked = Mock()
LotQt = Mock(return_value=source)
LotQt.link_to_transport.return_value = linked
Lot = Mock()
Lot.skip_quantity_consistency.return_value.__enter__ = Mock()
Lot.skip_quantity_consistency.return_value.__exit__ = Mock()
pool = Mock()
pool.get.side_effect = [LotQt, Lot]
wizard.records = [Mock(id=10000007)]
wizard.add = Mock(
shipment_in=shipment,
create_new_shipment=False,
lots=[Mock()],
unlink_remainder=True,
)
with patch('trytond.modules.purchase_trade.lot.Pool',
return_value=pool):
state = wizard.transition_adding()
self.assertEqual(state, 'end')
LotQt.link_to_transport.assert_called_once_with(
source, Decimal('125'), 'stock.shipment.in,40')
LotQt.add_physical_lots.assert_called_once_with(
linked, wizard.add.lots, True)
def test_add_physical_lot_creates_shipment_with_optional_bl(self):
'add physical lot creates and links a shipment with optional BL data'
wizard = lot_module.LotAdding()
shipment = Mock(id=40)
source = Mock(lot_quantity=Decimal('125'))
source.has_shipment.return_value = False
linked = Mock()
LotQt = Mock(return_value=source)
LotQt.link_to_transport.return_value = linked
ShipmentIn = Mock()
ShipmentIn.create.return_value = [shipment]
Lot = Mock()
Lot.skip_quantity_consistency.return_value.__enter__ = Mock()
Lot.skip_quantity_consistency.return_value.__exit__ = Mock()
pool = Mock()
pool.get.side_effect = [LotQt, ShipmentIn, Lot]
wizard.records = [Mock(id=10000007)]
wizard.add = Mock(
shipment_in=None,
create_new_shipment=True,
shipment_supplier=Mock(id=30),
transport_from_location=Mock(id=10),
transport_to_location=Mock(id=20),
shipment_bl_number='BL-2026-01',
shipment_bl_date=datetime.date(2026, 6, 22),
lots=[Mock()],
unlink_remainder=True,
)
with patch('trytond.modules.purchase_trade.lot.Pool',
return_value=pool):
state = wizard.transition_adding()
self.assertEqual(state, 'end')
ShipmentIn.create.assert_called_once_with([{
'supplier': 30,
'from_location': 10,
'to_location': 20,
'bl_number': 'BL-2026-01',
'bl_date': datetime.date(2026, 6, 22),
}])
LotQt.link_to_transport.assert_called_once_with(
source, Decimal('125'), 'stock.shipment.in,40')
LotQt.add_physical_lots.assert_called_once_with(
linked, wizard.add.lots, True)
@with_transaction()
def test_get_mtm_applies_component_ratio_as_percentage(self):
@@ -2604,6 +2779,18 @@ class PurchaseTradeTestCase(ModuleTestCase):
self.assertEqual(shipment.get_shipment_type(), 'inbound')
@with_transaction()
def test_shipment_in_location_domains_follow_trade_direction(self):
'shipment locations exclude invalid trade counterpart directions'
ShipmentIn = Pool().get('stock.shipment.in')
self.assertEqual(
ShipmentIn.from_location.domain,
[('type', '!=', 'customer')])
self.assertEqual(
ShipmentIn.to_location.domain,
[('type', '!=', 'supplier')])
@with_transaction()
def test_lot_report_shipment_type_uses_linked_shipment(self):
'lot report shipment type mirrors the linked shipment_in'

13
modules/purchase_trade/view/lot_add_start_form.xml
View File

@@ -7,6 +7,19 @@
<label name="unlink_remainder"/>
<field name="unlink_remainder"/>
<newline/>
<field name="transport_missing" invisible="1"/>
<label name="create_new_shipment"/>
<field name="create_new_shipment"/>
<label name="shipment_supplier"/>
<field name="shipment_supplier"/>
<label name="transport_from_location"/>
<field name="transport_from_location"/>
<label name="transport_to_location"/>
<field name="transport_to_location"/>
<label name="shipment_bl_number"/>
<field name="shipment_bl_number"/>
<label name="shipment_bl_date"/>
<field name="shipment_bl_date"/>
<label name="shipment_in"/>
<field name="shipment_in"/>
<label name="shipment_internal"/>