404 lines
14 KiB
Python
404 lines
14 KiB
Python
# 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; dates are set explicitly below
|
|
"cutoff_day": "0", # Monday
|
|
"pickup_day": "2", # Wednesday
|
|
"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_freezes_pickup_date_on_confirm(self):
|
|
"""Confirmed orders must keep the cycle pickup date (no next-week drift)."""
|
|
group_order = self._create_group_order(cutoff_in_past=True)
|
|
|
|
so = self._create_sale_order(group_order, self.member_1, self.consumer_group_1)
|
|
|
|
self.assertTrue(
|
|
group_order.pickup_date,
|
|
"Precondition failed: helper must provide a non-empty pickup_date",
|
|
)
|
|
|
|
# Simulate an inconsistent draft date (e.g. stale or shifted value)
|
|
wrong_pickup_date = group_order.pickup_date + timedelta(days=7)
|
|
so.write({"pickup_date": wrong_pickup_date})
|
|
|
|
self.assertEqual(so.state, "draft")
|
|
self.assertEqual(so.pickup_date, wrong_pickup_date)
|
|
|
|
expected_pickup_date = group_order.pickup_date
|
|
|
|
group_order._confirm_linked_sale_orders()
|
|
|
|
so.invalidate_recordset()
|
|
|
|
self.assertEqual(so.state, "sale")
|
|
self.assertEqual(
|
|
so.pickup_date,
|
|
expected_pickup_date,
|
|
"Cron should snapshot and preserve the current cycle pickup_date when confirming",
|
|
)
|
|
|
|
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",
|
|
)
|