Merge pull request #50846 from aerele/validate-product-bundle-stock-in-pos

erpnext/accounts/doctype/pos_invoice/pos_invoice.py

@@ -20,6 +20,11 @@ from erpnext.accounts.party import get_due_date, get_party_account
 from erpnext.controllers.queries import item_query as _item_query
 from erpnext.controllers.sales_and_purchase_return import get_sales_invoice_item_from_consolidated_invoice
 from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
+from erpnext.stock.stock_ledger import is_negative_stock_allowed
+
+
+class ProductBundleStockValidationError(frappe.ValidationError):
+	pass
 
 
 class POSInvoice(SalesInvoice):
@@ -395,32 +400,67 @@ class POSInvoice(SalesInvoice):
 
 		for d in self.get("items"):
 			if not d.serial_and_batch_bundle:
-				available_stock, is_stock_item, is_negative_stock_allowed = get_stock_availability(
+				if frappe.db.exists("Product Bundle", d.item_code):
-					d.item_code, d.warehouse
+					(
-				)
+						availability,
+						is_stock_item,
+						is_negative_stock_allowed,
+					) = get_product_bundle_stock_availability(d.item_code, d.warehouse, d.stock_qty)
+
+				else:
+					availability, is_stock_item, is_negative_stock_allowed = get_stock_availability(
+						d.item_code, d.warehouse
+					)
 
 				if is_negative_stock_allowed:
 					continue
 
-				item_code, warehouse, _qty = (
+				if isinstance(availability, list):
-					frappe.bold(d.item_code),
+					error_msgs = []
-					frappe.bold(d.warehouse),
+					for item in availability:
-					frappe.bold(d.qty),
+						if flt(item["available"]) < flt(item["required"]):
-				)
+							error_msgs.append(
-				if is_stock_item and flt(available_stock) <= 0:
+								_("<li>Packed Item {0}: Required {1}, Available {2}</li>").format(
-					frappe.throw(
+									frappe.bold(item["item_code"]),
-						_("Row #{}: Item Code: {} is not available under warehouse {}.").format(
+									frappe.bold(flt(item["required"], 2)),
-							d.idx, item_code, warehouse
+									frappe.bold(flt(item["available"], 2)),
-						),
+								)
-						title=_("Item Unavailable"),
+							)
-					)
+
-				elif is_stock_item and flt(available_stock) < flt(d.stock_qty):
+					if error_msgs:
-					frappe.throw(
+						frappe.throw(
-						_("Row #{}: Stock quantity not enough for Item Code: {} under warehouse {}.").format(
+							_(
-							d.idx, item_code, warehouse
+								"<b>Row #{0}:</b> Bundle {1} in warehouse {2} has insufficient packed items:<br><div style='margin-top: 15px;'><ul style='line-height: 0.8;'>{3}</ul></div>"
-						),
+							).format(
-						title=_("Item Unavailable"),
+								d.idx,
-					)
+								frappe.bold(d.item_code),
+								frappe.bold(d.warehouse),
+								"<br>".join(error_msgs),
+							),
+							title=_("Insufficient Stock for Product Bundle Items"),
+							exc=ProductBundleStockValidationError,
+						)
+
+				else:
+					item_code, warehouse = frappe.bold(d.item_code), frappe.bold(d.warehouse)
+					if is_stock_item and flt(availability) <= 0:
+						frappe.throw(
+							_("Row #{0}: Item {1} has no stock in warehouse {2}.").format(
+								d.idx, item_code, warehouse
+							),
+							title=_("Item Out of Stock"),
+						)
+					elif is_stock_item and flt(availability) < flt(d.stock_qty):
+						frappe.throw(
+							_("Row #{0}: Item {1} in warehouse {2}: Available {3}, Needed {4}.").format(
+								d.idx,
+								item_code,
+								warehouse,
+								frappe.bold(flt(availability, 2)),
+								frappe.bold(flt(d.stock_qty, 2)),
+							),
+							title=_("Insufficient Stock"),
+						)
 
 	def validate_is_pos_using_sales_invoice(self):
 		self.invoice_type_in_pos = frappe.db.get_single_value("POS Settings", "invoice_type")
@@ -858,8 +898,6 @@ class POSInvoice(SalesInvoice):
 
 @frappe.whitelist()
 def get_stock_availability(item_code, warehouse):
-	from erpnext.stock.stock_ledger import is_negative_stock_allowed
-
 	if frappe.db.get_value("Item", item_code, "is_stock_item"):
 		is_stock_item = True
 		bin_qty = get_bin_qty(item_code, warehouse)
@@ -876,6 +914,26 @@ def get_stock_availability(item_code, warehouse):
 			return 0, is_stock_item, False
 
 
+def get_product_bundle_stock_availability(item_code, warehouse, item_qty):
+	is_stock_item = True
+	bundle = frappe.get_doc("Product Bundle", item_code)
+	availabilities = []
+	for bundle_item in bundle.items:
+		if frappe.get_value("Item", bundle_item.item_code, "is_stock_item"):
+			bin_qty = get_bin_qty(bundle_item.item_code, warehouse)
+			reserved_qty = get_pos_reserved_qty(bundle_item.item_code, warehouse)
+			available = bin_qty - reserved_qty
+			availabilities.append(
+				{
+					"item_code": bundle_item.item_code,
+					"required": bundle_item.qty * item_qty,
+					"available": available,
+				}
+			)
+
+	return availabilities, is_stock_item, is_negative_stock_allowed(item_code=item_code)
+
+
 def get_bundle_availability(bundle_item_code, warehouse):
 	product_bundle = frappe.get_doc("Product Bundle", bundle_item_code)
 

erpnext/accounts/doctype/pos_invoice/test_pos_invoice.py

@@ -1024,6 +1024,84 @@ class TestPOSInvoice(IntegrationTestCase):
 			frappe.db.rollback(save_point="before_test_delivered_serial_no_case")
 			frappe.set_user("Administrator")
 
+	def test_bundle_stock_availability_validation(self):
+		from erpnext.accounts.doctype.pos_invoice.pos_invoice import ProductBundleStockValidationError
+		from erpnext.accounts.doctype.pos_invoice_merge_log.test_pos_invoice_merge_log import (
+			init_user_and_profile,
+		)
+		from erpnext.selling.doctype.product_bundle.test_product_bundle import make_product_bundle
+		from erpnext.stock.doctype.item.test_item import create_item
+
+		init_user_and_profile()
+
+		frappe.set_user("Administrator")
+
+		warehouse = "_Test Warehouse - _TC"
+		company = "_Test Company"
+
+		# Create stock sub-items
+		sub_item_a = "_Test Bundle SubA"
+		if not frappe.db.exists("Item", sub_item_a):
+			create_item(
+				item_code=sub_item_a,
+				is_stock_item=1,
+			)
+
+		sub_item_b = "_Test Bundle SubB"
+		if not frappe.db.exists("Item", sub_item_b):
+			create_item(
+				item_code=sub_item_b,
+				is_stock_item=1,
+			)
+
+		# Add initial stock: SubA=5, SubB=2
+		make_stock_entry(item_code=sub_item_a, target=warehouse, qty=5, company=company)
+		make_stock_entry(item_code=sub_item_b, target=warehouse, qty=2, company=company)
+
+		# Create Product Bundle: Test Bundle (SubA x2 + SubB x1)
+		bundle_item = "_Test Bundle"
+		if not frappe.db.exists("Item", bundle_item):
+			create_item(
+				item_code=bundle_item,
+				is_stock_item=0,
+			)
+
+		if not frappe.db.exists("Product Bundle", bundle_item):
+			make_product_bundle(parent=bundle_item, items=[sub_item_a, sub_item_b])
+
+		# Test Case 1: Sufficient stock (bundle qty=1: requires SubA=2 (<=5), SubB=1 (<=2)) -> No error
+		pos_inv_sufficient = create_pos_invoice(
+			item=bundle_item,
+			qty=1,
+			rate=100,
+			warehouse=warehouse,
+			pos_profile=self.pos_profile.name,
+			do_not_save=1,
+		)
+		pos_inv_sufficient.append("payments", {"mode_of_payment": "Cash", "amount": 100, "default": 1})
+		pos_inv_sufficient.insert()
+		pos_inv_sufficient.submit()
+
+		pos_inv_sufficient.cancel()
+		pos_inv_sufficient.delete()
+
+		# Test Case 2: Insufficient stock (reduce SubB to 1, bundle qty=2: requires SubB=2 >1) -> Error with details
+		make_stock_entry(item_code=sub_item_b, from_warehouse=warehouse, qty=1, company=company)
+
+		pos_inv_insufficient = create_pos_invoice(
+			item=bundle_item,
+			qty=2,
+			rate=100,
+			warehouse=warehouse,
+			pos_profile=self.pos_profile.name,
+			do_not_save=1,
+		)
+		pos_inv_insufficient.append("payments", {"mode_of_payment": "Cash", "amount": 200, "default": 1})
+		pos_inv_insufficient.save()
+		self.assertRaises(ProductBundleStockValidationError, pos_inv_insufficient.submit)
+
+		frappe.set_user("test@example.com")
+
 
 def create_pos_invoice(**args):
 	args = frappe._dict(args)