[ADD] website_sale_aplicoop: create picking batches after cutoff

- Add stock_picking_batch dependency to manifest
- Add cutoff date validation in _confirm_linked_sale_orders()
- Create _create_picking_batches_for_sale_orders() method
- Group pickings by consumer_group_id into separate batches
- Set batch scheduled_date from group order pickup_date
- Add test_cron_picking_batch.py with 7 tests covering:
  - Skip orders before cutoff
  - Confirm orders after cutoff
  - Separate batches per consumer group
  - Same group orders in same batch
  - Batch has scheduled_date
  - No duplicate batches on re-run
  - Closed group orders not processed
This commit is contained in:
snt 2026-03-06 15:45:12 +01:00
parent c3173a32c9
commit e2ced75ecd
4 changed files with 455 additions and 11 deletions

View file

@ -16,6 +16,7 @@
"product",
"sale",
"stock",
"stock_picking_batch",
"account",
"product_get_price_helper",
],

View file

@ -688,8 +688,23 @@ class GroupOrder(models.Model):
This is triggered by the daily cron so that weekly orders generated
from the website are confirmed automatically once dates are refreshed.
After confirmation, creates picking batches grouped by consumer group.
Only confirms orders if the cutoff date has already passed.
"""
self.ensure_one()
# Only confirm if cutoff date has passed
today = fields.Date.today()
if self.cutoff_date and self.cutoff_date >= today:
_logger.info(
"Cron: Skipping group order %s (%s) - cutoff date %s not yet passed",
self.id,
self.name,
self.cutoff_date,
)
return
SaleOrder = self.env["sale.order"].sudo()
sale_orders = SaleOrder.search(
[
@ -715,9 +730,65 @@ class GroupOrder(models.Model):
try:
sale_orders.action_confirm()
# Create picking batches after confirmation
self._create_picking_batches_for_sale_orders(sale_orders)
except Exception:
_logger.exception(
"Cron: Error confirming sale orders for group order %s (%s)",
self.id,
self.name,
)
def _create_picking_batches_for_sale_orders(self, sale_orders):
"""Create stock.picking.batch grouped by consumer_group_id.
Args:
sale_orders: Recordset of confirmed sale.order
"""
self.ensure_one()
StockPickingBatch = self.env["stock.picking.batch"].sudo()
# Group sale orders by consumer_group_id
groups = {}
for so in sale_orders:
group_id = so.consumer_group_id.id or False
if group_id not in groups:
groups[group_id] = self.env["sale.order"]
groups[group_id] |= so
for consumer_group_id, group_sale_orders in groups.items():
# Get pickings without batch
pickings = group_sale_orders.picking_ids.filtered(
lambda p: p.state not in ("done", "cancel") and not p.batch_id
)
if not pickings:
continue
# Get consumer group name for batch description
consumer_group = self.env["res.partner"].browse(consumer_group_id)
batch_desc = (
f"{self.name} - {consumer_group.name}" if consumer_group else self.name
)
# Create the batch
batch = StockPickingBatch.create(
{
"description": batch_desc,
"company_id": self.company_id.id,
"picking_type_id": pickings[0].picking_type_id.id,
"scheduled_date": self.pickup_date,
}
)
# Assign pickings to the batch
pickings.write({"batch_id": batch.id})
_logger.info(
"Cron: Created batch %s with %d pickings for group order %s, "
"consumer group %s",
batch.name,
len(pickings),
self.name,
consumer_group.name if consumer_group else "N/A",
)

View file

@ -1,14 +1,15 @@
# Copyright 2025 Criptomart
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl)
from . import test_group_order
from . import test_res_partner
from . import test_product_extension
from . import test_eskaera_shop
from . import test_templates_rendering
from . import test_record_rules
from . import test_multi_company
from . import test_save_order_endpoints
from . import test_date_calculations
from . import test_pricing_with_pricelist
from . import test_portal_sale_order_creation
from . import test_group_order # noqa: F401
from . import test_res_partner # noqa: F401
from . import test_product_extension # noqa: F401
from . import test_eskaera_shop # noqa: F401
from . import test_templates_rendering # noqa: F401
from . import test_record_rules # noqa: F401
from . import test_multi_company # noqa: F401
from . import test_save_order_endpoints # noqa: F401
from . import test_date_calculations # noqa: F401
from . import test_pricing_with_pricelist # noqa: F401
from . import test_portal_sale_order_creation # noqa: F401
from . import test_cron_picking_batch # noqa: F401

View file

@ -0,0 +1,371 @@
# Copyright 2026 Criptomart
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl)
from datetime import timedelta
from odoo import fields
from odoo.tests.common import TransactionCase
from odoo.tests.common import tagged
@tagged("post_install", "cron_picking_batch")
class TestCronPickingBatch(TransactionCase):
"""Test suite for cron jobs that confirm sale orders and create picking batches."""
@classmethod
def setUpClass(cls):
super().setUpClass()
# Create consumer groups
cls.consumer_group_1 = cls.env["res.partner"].create(
{
"name": "Consumer Group 1",
"is_company": True,
"is_group": True,
"email": "group1@test.com",
}
)
cls.consumer_group_2 = cls.env["res.partner"].create(
{
"name": "Consumer Group 2",
"is_company": True,
"is_group": True,
"email": "group2@test.com",
}
)
# Create test members
cls.member_1 = cls.env["res.partner"].create(
{
"name": "Member 1",
"email": "member1@test.com",
"parent_id": cls.consumer_group_1.id,
}
)
cls.member_2 = cls.env["res.partner"].create(
{
"name": "Member 2",
"email": "member2@test.com",
"parent_id": cls.consumer_group_2.id,
}
)
# Create a test product (storable to generate pickings)
cls.product = cls.env["product.product"].create(
{
"name": "Test Product",
"is_storable": True, # Odoo 18: storable products generate pickings
"list_price": 10.0,
}
)
def _create_group_order(self, cutoff_in_past=False, state="open"):
"""Create a group order with cutoff date in past or future.
Args:
cutoff_in_past: If True, cutoff_date will be yesterday (past).
If False, cutoff_date will be tomorrow (future).
state: State of the group order
"""
today = fields.Date.today()
# Create with basic config first
order = self.env["group.order"].create(
{
"name": f"Test Group Order {'past' if cutoff_in_past else 'future'}",
"group_ids": [
(6, 0, [self.consumer_group_1.id, self.consumer_group_2.id])
],
"period": "once", # One-time order to avoid date recalculation
"state": state,
}
)
# Directly set cutoff_date and pickup_date (bypass computed logic)
if cutoff_in_past:
cutoff_date = today - timedelta(days=1) # Yesterday
pickup_date = today + timedelta(days=1) # Tomorrow
else:
cutoff_date = today + timedelta(days=2) # Day after tomorrow
pickup_date = today + timedelta(days=4) # 4 days from now
# Write directly to stored computed fields
order.write(
{
"cutoff_date": cutoff_date,
"pickup_date": pickup_date,
}
)
return order
def _create_sale_order(self, group_order, partner, consumer_group):
"""Create a draft sale order linked to the group order."""
return self.env["sale.order"].create(
{
"partner_id": partner.id,
"group_order_id": group_order.id,
"consumer_group_id": consumer_group.id,
"pickup_date": group_order.pickup_date,
"order_line": [
(
0,
0,
{
"product_id": self.product.id,
"product_uom_qty": 1,
"price_unit": 10.0,
},
)
],
}
)
def test_cron_skips_orders_before_cutoff(self):
"""Test that cron does NOT confirm orders if cutoff date has not passed."""
# Create group order with cutoff in future
group_order = self._create_group_order(cutoff_in_past=False)
# Create draft sale orders
so1 = self._create_sale_order(group_order, self.member_1, self.consumer_group_1)
so2 = self._create_sale_order(group_order, self.member_2, self.consumer_group_2)
self.assertEqual(so1.state, "draft")
self.assertEqual(so2.state, "draft")
# Verify cutoff is in future
today = fields.Date.today()
self.assertGreater(
group_order.cutoff_date,
today,
"Cutoff date should be in the future for this test",
)
# Call the confirmation method directly (not full cron to avoid date recalc)
group_order._confirm_linked_sale_orders()
# Sale orders should still be draft (cutoff not passed)
self.assertEqual(
so1.state,
"draft",
"Sale order should remain draft - cutoff date not yet passed",
)
self.assertEqual(
so2.state,
"draft",
"Sale order should remain draft - cutoff date not yet passed",
)
def test_cron_confirms_orders_after_cutoff(self):
"""Test that cron confirms orders when cutoff date has passed."""
# Create group order with cutoff yesterday (past)
group_order = self._create_group_order(cutoff_in_past=True)
# Create draft sale orders
so1 = self._create_sale_order(group_order, self.member_1, self.consumer_group_1)
so2 = self._create_sale_order(group_order, self.member_2, self.consumer_group_2)
self.assertEqual(so1.state, "draft")
self.assertEqual(so2.state, "draft")
# Verify cutoff is in past
today = fields.Date.today()
self.assertLess(
group_order.cutoff_date,
today,
"Cutoff date should be in the past for this test",
)
# Call the confirmation method directly
group_order._confirm_linked_sale_orders()
# Refresh records
so1.invalidate_recordset()
so2.invalidate_recordset()
# Sale orders should be confirmed (cutoff passed)
self.assertEqual(
so1.state,
"sale",
"Sale order should be confirmed - cutoff date has passed",
)
self.assertEqual(
so2.state,
"sale",
"Sale order should be confirmed - cutoff date has passed",
)
def test_cron_creates_picking_batch_per_consumer_group(self):
"""Test that cron creates separate picking batches per consumer group."""
# Create group order with cutoff yesterday (past)
group_order = self._create_group_order(cutoff_in_past=True)
# Create draft sale orders for different consumer groups
so1 = self._create_sale_order(group_order, self.member_1, self.consumer_group_1)
so2 = self._create_sale_order(group_order, self.member_2, self.consumer_group_2)
# Call the confirmation method directly
group_order._confirm_linked_sale_orders()
# Refresh records
so1.invalidate_recordset()
so2.invalidate_recordset()
# Check that pickings were created
self.assertTrue(so1.picking_ids, "Sale order 1 should have pickings")
self.assertTrue(so2.picking_ids, "Sale order 2 should have pickings")
# Check that pickings have batch_id assigned
for picking in so1.picking_ids:
self.assertTrue(
picking.batch_id,
"Picking from SO1 should be assigned to a batch",
)
for picking in so2.picking_ids:
self.assertTrue(
picking.batch_id,
"Picking from SO2 should be assigned to a batch",
)
# Check that batches are different (one per consumer group)
batch_1 = so1.picking_ids[0].batch_id
batch_2 = so2.picking_ids[0].batch_id
self.assertNotEqual(
batch_1.id,
batch_2.id,
"Different consumer groups should have different batches",
)
# Check batch descriptions contain consumer group names
self.assertIn(
self.consumer_group_1.name,
batch_1.description,
"Batch 1 description should include consumer group 1 name",
)
self.assertIn(
self.consumer_group_2.name,
batch_2.description,
"Batch 2 description should include consumer group 2 name",
)
def test_cron_same_consumer_group_same_batch(self):
"""Test that orders from same consumer group go to same batch."""
# Create group order with cutoff yesterday (past)
group_order = self._create_group_order(cutoff_in_past=True)
# Create another member in the same group
member_1b = self.env["res.partner"].create(
{
"name": "Member 1B",
"email": "member1b@test.com",
"parent_id": self.consumer_group_1.id,
}
)
# Create two sale orders from same consumer group
so1 = self._create_sale_order(group_order, self.member_1, self.consumer_group_1)
so2 = self._create_sale_order(group_order, member_1b, self.consumer_group_1)
# Call the confirmation method directly
group_order._confirm_linked_sale_orders()
# Refresh records
so1.invalidate_recordset()
so2.invalidate_recordset()
# Check that both pickings are in the same batch
batch_1 = so1.picking_ids[0].batch_id
batch_2 = so2.picking_ids[0].batch_id
self.assertEqual(
batch_1.id,
batch_2.id,
"Same consumer group should have same batch",
)
# Check batch has 2 pickings
self.assertEqual(
len(batch_1.picking_ids),
2,
"Batch should contain 2 pickings from same consumer group",
)
def test_cron_batch_scheduled_date(self):
"""Test that batch has a scheduled_date set."""
# Create group order with cutoff yesterday (past)
group_order = self._create_group_order(cutoff_in_past=True)
# Create a sale order
so = self._create_sale_order(group_order, self.member_1, self.consumer_group_1)
# Call the confirmation method directly
group_order._confirm_linked_sale_orders()
# Refresh records
so.invalidate_recordset()
# Check batch exists and has scheduled_date
self.assertTrue(so.picking_ids, "Sale order should have pickings")
batch = so.picking_ids[0].batch_id
self.assertTrue(batch, "Picking should have a batch")
# scheduled_date should be set (not False/None)
self.assertTrue(
batch.scheduled_date,
"Batch should have a scheduled_date set",
)
def test_cron_does_not_duplicate_batches(self):
"""Test that running cron twice does not create duplicate batches."""
# Create group order with cutoff yesterday (past)
group_order = self._create_group_order(cutoff_in_past=True)
# Create a sale order
so = self._create_sale_order(group_order, self.member_1, self.consumer_group_1)
# Call first time
group_order._confirm_linked_sale_orders()
# Refresh records
so.invalidate_recordset()
self.assertTrue(so.picking_ids, "Sale order should have pickings")
batch_first = so.picking_ids[0].batch_id
batch_count_first = self.env["stock.picking.batch"].search_count([])
# Call second time
group_order._confirm_linked_sale_orders()
batch_second = so.picking_ids[0].batch_id
batch_count_second = self.env["stock.picking.batch"].search_count([])
# Should be same batch, no duplicates
self.assertEqual(
batch_first.id,
batch_second.id,
"Batch should not change after second cron execution",
)
self.assertEqual(
batch_count_first,
batch_count_second,
"No new batches should be created on second cron execution",
)
def test_cron_closed_group_order_not_processed(self):
"""Test that closed group orders are not processed by cron."""
# Create group order with cutoff yesterday but state=closed
group_order = self._create_group_order(cutoff_in_past=True, state="closed")
# Create draft sale order
so = self._create_sale_order(group_order, self.member_1, self.consumer_group_1)
# Execute full cron (which only processes draft/open orders)
self.env["group.order"]._cron_update_dates()
# Refresh
so.invalidate_recordset()
# Sale order should still be draft (closed group orders not processed)
self.assertEqual(
so.state,
"draft",
"Sale order should remain draft - group order is closed",
)