# Copyright 2026 Criptomart # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). from unittest.mock import patch from odoo.exceptions import UserError from odoo.tests import TransactionCase from odoo.tests import tagged @tagged("-at_install", "post_install") class TestBatchSummary(TransactionCase): @classmethod def setUpClass(cls): super().setUpClass() cls.uom_unit = cls.env.ref("uom.product_uom_unit") cls.uom_dozen = cls.env.ref("uom.product_uom_dozen") cls.location_src = cls.env.ref("stock.stock_location_stock") cls.location_dest = cls.env.ref("stock.stock_location_customers") cls.picking_type = cls.env.ref("stock.picking_type_out") cls.product = cls.env["product.product"].create( { "name": "Test Product", "uom_id": cls.uom_unit.id, "uom_po_id": cls.uom_unit.id, } ) cls.product_2 = cls.env["product.product"].create( { "name": "Test Product 2", "uom_id": cls.uom_unit.id, "uom_po_id": cls.uom_unit.id, } ) cls.batch = cls.env["stock.picking.batch"].create( {"name": "Batch Test", "picking_type_id": cls.picking_type.id} ) cls.picking1 = cls.env["stock.picking"].create( { "picking_type_id": cls.picking_type.id, "location_id": cls.location_src.id, "location_dest_id": cls.location_dest.id, "batch_id": cls.batch.id, } ) cls.picking2 = cls.env["stock.picking"].create( { "picking_type_id": cls.picking_type.id, "location_id": cls.location_src.id, "location_dest_id": cls.location_dest.id, "batch_id": cls.batch.id, } ) # Helpers def _add_move(self, picking, qty_demanded, qty_done, uom): move = self.env["stock.move"].create( { "name": self.product.name, "product_id": self.product.id, "product_uom_qty": qty_demanded, "product_uom": uom.id, "picking_id": picking.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, } ) if qty_done: self.env["stock.move.line"].create( { "move_id": move.id, "picking_id": picking.id, "product_id": self.product.id, "product_uom_id": self.product.uom_id.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, "quantity": qty_done, } ) return move def _recompute_batch(self): self.batch.invalidate_recordset() self.batch._compute_summary_line_ids() def _create_batch_with_pickings(self): batch = self.env["stock.picking.batch"].create( {"name": "Batch Flow", "picking_type_id": self.picking_type.id} ) picking = self.env["stock.picking"].create( { "picking_type_id": self.picking_type.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, } ) move = self.env["stock.move"].create( { "name": self.product.name, "product_id": self.product.id, "product_uom_qty": 2.0, "product_uom": self.uom_unit.id, "picking_id": picking.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, } ) self.env["stock.move.line"].create( { "move_id": move.id, "picking_id": picking.id, "product_id": self.product.id, "product_uom_id": self.uom_unit.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, "quantity": 2.0, } ) picking.batch_id = batch.id return batch def _create_partial_batch(self, qty_demanded=2.0, qty_done=1.0, extra_move=False): batch = self.env["stock.picking.batch"].create( {"name": "Batch Partial", "picking_type_id": self.picking_type.id} ) batch.picking_type_id.create_backorder = "ask" picking = self.env["stock.picking"].create( { "picking_type_id": self.picking_type.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, "batch_id": batch.id, } ) move = self.env["stock.move"].create( { "name": self.product.name, "product_id": self.product.id, "product_uom_qty": qty_demanded, "product_uom": self.uom_unit.id, "picking_id": picking.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, } ) self.env["stock.move.line"].create( { "move_id": move.id, "picking_id": picking.id, "product_id": self.product.id, "product_uom_id": self.uom_unit.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, "quantity": qty_done, } ) if extra_move: self.env["stock.move"].create( { "name": self.product_2.name, "product_id": self.product_2.id, "product_uom_qty": 1.0, "product_uom": self.uom_unit.id, "picking_id": picking.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, } ) batch.action_confirm() batch._compute_summary_line_ids() return batch def _set_batch_restriction_config(self, **vals): company = self.env.company previous = {key: company[key] for key in vals} company.write(vals) self.addCleanup(lambda: company.write(previous)) # Tests def test_totals_and_pending_with_conversion(self): """Totals aggregate per product with UoM conversion and pending.""" # demand 12 units, done 5 units self._add_move(self.picking1, qty_demanded=12, qty_done=5, uom=self.uom_unit) # demand 2 dozens (24 units), done 6 units self._add_move(self.picking2, qty_demanded=2, qty_done=6, uom=self.uom_dozen) self._recompute_batch() self.assertEqual(len(self.batch.summary_line_ids), 1) line = self.batch.summary_line_ids self.assertEqual(line.product_id, self.product) self.assertEqual(line.product_uom_id, self.uom_unit) self.assertAlmostEqual(line.qty_demanded, 36.0) self.assertAlmostEqual(line.qty_done, 11.0) self.assertAlmostEqual(line.qty_pending, 25.0) def test_collected_flag_preserved_on_recompute(self): """Collected stays checked after totals change.""" self._add_move(self.picking1, qty_demanded=1, qty_done=1, uom=self.uom_unit) self._recompute_batch() line = self.batch.summary_line_ids line.is_collected = True # Add more demand/done to trigger an update self._add_move(self.picking1, qty_demanded=3, qty_done=2, uom=self.uom_unit) self._recompute_batch() self.assertTrue(self.batch.summary_line_ids.is_collected) def test_cancelled_moves_are_ignored(self): """Cancelled moves do not count in the summary and lines are removed.""" move1 = self._add_move( self.picking1, qty_demanded=4, qty_done=2, uom=self.uom_unit ) move2 = self._add_move( self.picking2, qty_demanded=6, qty_done=3, uom=self.uom_unit ) self._recompute_batch() self.assertEqual(len(self.batch.summary_line_ids), 1) move1.write({"state": "cancel"}) move2.write({"state": "cancel"}) self._recompute_batch() self.assertFalse(self.batch.summary_line_ids) def test_no_required_product_error_on_confirm(self): """Confirming batch with pickings must not create summary lines without product.""" batch = self._create_batch_with_pickings() # Trigger compute via confirm flow batch.action_confirm() self.assertTrue(batch.summary_line_ids) self.assertFalse( batch.summary_line_ids.filtered(lambda line: not line.product_id) ) def test_empty_batch_add_pickings_then_confirm(self): """Create empty draft batch, add multiple pickings, then confirm.""" # 1. Create empty batch in draft state batch = self.env["stock.picking.batch"].create( {"name": "Batch Empty Start", "picking_type_id": self.picking_type.id} ) self.assertEqual(batch.state, "draft") self.assertFalse(batch.picking_ids) self.assertFalse(batch.summary_line_ids) # 2. Create pickings with moves (without batch assignment) picking_a = self.env["stock.picking"].create( { "picking_type_id": self.picking_type.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, } ) self.env["stock.move"].create( { "name": self.product.name, "product_id": self.product.id, "product_uom_qty": 5.0, "product_uom": self.uom_unit.id, "picking_id": picking_a.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, } ) picking_b = self.env["stock.picking"].create( { "picking_type_id": self.picking_type.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, } ) self.env["stock.move"].create( { "name": self.product_2.name, "product_id": self.product_2.id, "product_uom_qty": 10.0, "product_uom": self.uom_unit.id, "picking_id": picking_b.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, } ) picking_c = self.env["stock.picking"].create( { "picking_type_id": self.picking_type.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, } ) self.env["stock.move"].create( { "name": self.product.name, "product_id": self.product.id, "product_uom_qty": 3.0, "product_uom": self.uom_unit.id, "picking_id": picking_c.id, "location_id": self.location_src.id, "location_dest_id": self.location_dest.id, } ) # 3. Add pickings to the batch picking_a.batch_id = batch picking_b.batch_id = batch picking_c.batch_id = batch self.assertEqual(len(batch.picking_ids), 3) # 4. Confirm the batch — this should not raise product_id required error batch.action_confirm() self.assertEqual(batch.state, "in_progress") # 5. Verify summary lines are correct self.assertTrue(batch.summary_line_ids) self.assertFalse( batch.summary_line_ids.filtered(lambda line: not line.product_id) ) # Two products expected products_in_summary = batch.summary_line_ids.mapped("product_id") self.assertIn(self.product, products_in_summary) self.assertIn(self.product_2, products_in_summary) # Check aggregated quantities line_product1 = batch.summary_line_ids.filtered( lambda line: line.product_id == self.product ) self.assertAlmostEqual(line_product1.qty_demanded, 8.0) # 5 + 3 line_product2 = batch.summary_line_ids.filtered( lambda line: line.product_id == self.product_2 ) self.assertAlmostEqual(line_product2.qty_demanded, 10.0) def test_done_requires_detailed_lines_collected_without_backorder(self): """Full validation blocks when detailed operation lines are unchecked.""" batch = self._create_batch_with_pickings() batch.action_confirm() batch._compute_summary_line_ids() detail_lines = batch.move_line_ids.filtered(lambda line: line.quantity > 0) self.assertTrue(detail_lines) self.assertFalse(any(detail_lines.mapped("is_collected"))) with self.assertRaises(UserError): batch.action_done() def test_done_ignores_product_summary_checkbox(self): """Product Summary checkbox is informative and must not block validation.""" batch = self._create_batch_with_pickings() batch.action_confirm() batch._compute_summary_line_ids() # Keep Product Summary unchecked on purpose self.assertFalse(any(batch.summary_line_ids.mapped("is_collected"))) # Detailed lines drive validation batch.move_line_ids.filtered(lambda line: line.quantity > 0).write( {"is_collected": True} ) # Should not raise batch.action_done() def test_partial_validation_waits_for_backorder_wizard_before_blocking(self): """Partial batch validation must open the backorder wizard first.""" batch = self._create_partial_batch() with patch.object( type(self.env["stock.picking"]), "_check_backorder", autospec=True, return_value=batch.picking_ids, ): action = batch.action_done() self.assertIsInstance(action, dict) self.assertEqual(action.get("res_model"), "stock.backorder.confirmation") def test_partial_validation_checks_only_processed_detailed_lines(self): """Only processed detailed lines must be required for collected.""" batch = self._create_partial_batch(extra_move=True) # Ensure product_2 remains non-processed in this scenario move_product_2 = batch.move_ids.filtered( lambda move: move.product_id == self.product_2 ) move_product_2.write({"quantity": 0.0}) move_product_2.move_line_ids.write({"quantity": 0.0}) with self.assertRaisesRegex(UserError, self.product.display_name) as err: batch._check_all_products_collected(batch.picking_ids) self.assertNotIn(self.product_2.display_name, str(err.exception)) def test_check_all_products_collected_passes_when_all_checked(self): """Collected validation helper must pass when all lines are checked.""" batch = self._create_batch_with_pickings() batch.action_confirm() batch._compute_summary_line_ids() batch.move_line_ids.filtered(lambda line: line.quantity > 0).write( {"is_collected": True} ) # Should not raise batch._check_all_products_collected() def test_summary_restriction_blocks_when_enabled(self): """Product Summary restriction can be enabled independently.""" self._set_batch_restriction_config( batch_summary_restriction_enabled=True, batch_summary_restriction_scope="processed", batch_detailed_restriction_enabled=False, ) batch = self._create_batch_with_pickings() batch.action_confirm() batch._compute_summary_line_ids() # Detailed lines are collected, but Product Summary stays unchecked batch.move_line_ids.filtered(lambda line: line.quantity > 0).write( {"is_collected": True} ) with self.assertRaisesRegex(UserError, "product summary"): batch.action_done() def test_detailed_scope_all_blocks_unprocessed_lines(self): """Detailed scope 'all' must include non-processed detailed lines.""" self._set_batch_restriction_config( batch_summary_restriction_enabled=False, batch_detailed_restriction_enabled=True, batch_detailed_restriction_scope="all", ) batch = self._create_partial_batch(extra_move=True) with self.assertRaisesRegex(UserError, self.product_2.display_name): batch._check_all_products_collected(batch.picking_ids)