Merge pull request #29456 from marination/product-bundle-packing-list-logic
fix: Incorrect packing list for recurring items & code cleanup
This commit is contained in:
Marica
GitHub
commit
79ab8e6459
@ -441,7 +441,7 @@ erpnext.buying.get_items_from_product_bundle = function(frm) {
|
|||||||
type: "GET",
|
type: "GET",
|
||||||
method: "erpnext.stock.doctype.packed_item.packed_item.get_items_from_product_bundle",
|
method: "erpnext.stock.doctype.packed_item.packed_item.get_items_from_product_bundle",
|
||||||
args: {
|
args: {
|
||||||
args: {
|
row: {
|
||||||
item_code: args.product_bundle,
|
item_code: args.product_bundle,
|
||||||
quantity: args.quantity,
|
quantity: args.quantity,
|
||||||
parenttype: frm.doc.doctype,
|
parenttype: frm.doc.doctype,
|
||||||
|
|||||||
@ -218,8 +218,6 @@
|
|||||||
"label": "Conversion Factor"
|
"label": "Conversion Factor"
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
"fetch_from": "item_code.valuation_rate",
|
|
||||||
"fetch_if_empty": 1,
|
|
||||||
"fieldname": "rate",
|
"fieldname": "rate",
|
||||||
"fieldtype": "Currency",
|
"fieldtype": "Currency",
|
||||||
"in_list_view": 1,
|
"in_list_view": 1,
|
||||||
@ -232,7 +230,7 @@
|
|||||||
"index_web_pages_for_search": 1,
|
"index_web_pages_for_search": 1,
|
||||||
"istable": 1,
|
"istable": 1,
|
||||||
"links": [],
|
"links": [],
|
||||||
"modified": "2021-09-01 15:10:29.646399",
|
"modified": "2022-01-28 16:03:30.780111",
|
||||||
"modified_by": "Administrator",
|
"modified_by": "Administrator",
|
||||||
"module": "Stock",
|
"module": "Stock",
|
||||||
"name": "Packed Item",
|
"name": "Packed Item",
|
||||||
@ -240,5 +238,6 @@
|
|||||||
"permissions": [],
|
"permissions": [],
|
||||||
"sort_field": "modified",
|
"sort_field": "modified",
|
||||||
"sort_order": "DESC",
|
"sort_order": "DESC",
|
||||||
|
"states": [],
|
||||||
"track_changes": 1
|
"track_changes": 1
|
||||||
}
|
}
|
||||||
@ -8,187 +8,250 @@ import json
|
|||||||
|
|
||||||
import frappe
|
import frappe
|
||||||
from frappe.model.document import Document
|
from frappe.model.document import Document
|
||||||
from frappe.utils import cstr, flt
|
from frappe.utils import flt
|
||||||
|
|
||||||
from erpnext.stock.get_item_details import get_item_details
|
from erpnext.stock.get_item_details import get_item_details, get_price_list_rate
|
||||||
|
|
||||||
|
|
||||||
class PackedItem(Document):
|
class PackedItem(Document):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def get_product_bundle_items(item_code):
|
|
||||||
return frappe.db.sql("""select t1.item_code, t1.qty, t1.uom, t1.description
|
|
||||||
from `tabProduct Bundle Item` t1, `tabProduct Bundle` t2
|
|
||||||
where t2.new_item_code=%s and t1.parent = t2.name order by t1.idx""", item_code, as_dict=1)
|
|
||||||
|
|
||||||
def get_packing_item_details(item, company):
|
|
||||||
return frappe.db.sql("""
|
|
||||||
select i.item_name, i.is_stock_item, i.description, i.stock_uom, id.default_warehouse
|
|
||||||
from `tabItem` i LEFT JOIN `tabItem Default` id ON id.parent=i.name and id.company=%s
|
|
||||||
where i.name = %s""",
|
|
||||||
(company, item), as_dict = 1)[0]
|
|
||||||
|
|
||||||
def get_bin_qty(item, warehouse):
|
|
||||||
det = frappe.db.sql("""select actual_qty, projected_qty from `tabBin`
|
|
||||||
where item_code = %s and warehouse = %s""", (item, warehouse), as_dict = 1)
|
|
||||||
return det and det[0] or frappe._dict()
|
|
||||||
|
|
||||||
def update_packing_list_item(doc, packing_item_code, qty, main_item_row, description):
|
|
||||||
if doc.amended_from:
|
|
||||||
old_packed_items_map = get_old_packed_item_details(doc.packed_items)
|
|
||||||
else:
|
|
||||||
old_packed_items_map = False
|
|
||||||
item = get_packing_item_details(packing_item_code, doc.company)
|
|
||||||
|
|
||||||
# check if exists
|
|
||||||
exists = 0
|
|
||||||
for d in doc.get("packed_items"):
|
|
||||||
if d.parent_item == main_item_row.item_code and d.item_code == packing_item_code:
|
|
||||||
if d.parent_detail_docname != main_item_row.name:
|
|
||||||
d.parent_detail_docname = main_item_row.name
|
|
||||||
|
|
||||||
pi, exists = d, 1
|
|
||||||
break
|
|
||||||
|
|
||||||
if not exists:
|
|
||||||
pi = doc.append('packed_items', {})
|
|
||||||
|
|
||||||
pi.parent_item = main_item_row.item_code
|
|
||||||
pi.item_code = packing_item_code
|
|
||||||
pi.item_name = item.item_name
|
|
||||||
pi.parent_detail_docname = main_item_row.name
|
|
||||||
pi.uom = item.stock_uom
|
|
||||||
pi.qty = flt(qty)
|
|
||||||
pi.conversion_factor = main_item_row.conversion_factor
|
|
||||||
if description and not pi.description:
|
|
||||||
pi.description = description
|
|
||||||
if not pi.warehouse and not doc.amended_from:
|
|
||||||
pi.warehouse = (main_item_row.warehouse if ((doc.get('is_pos') or item.is_stock_item \
|
|
||||||
or not item.default_warehouse) and main_item_row.warehouse) else item.default_warehouse)
|
|
||||||
if not pi.batch_no and not doc.amended_from:
|
|
||||||
pi.batch_no = cstr(main_item_row.get("batch_no"))
|
|
||||||
if not pi.target_warehouse:
|
|
||||||
pi.target_warehouse = main_item_row.get("target_warehouse")
|
|
||||||
bin = get_bin_qty(packing_item_code, pi.warehouse)
|
|
||||||
pi.actual_qty = flt(bin.get("actual_qty"))
|
|
||||||
pi.projected_qty = flt(bin.get("projected_qty"))
|
|
||||||
if old_packed_items_map and old_packed_items_map.get((packing_item_code, main_item_row.item_code)):
|
|
||||||
pi.batch_no = old_packed_items_map.get((packing_item_code, main_item_row.item_code))[0].batch_no
|
|
||||||
pi.serial_no = old_packed_items_map.get((packing_item_code, main_item_row.item_code))[0].serial_no
|
|
||||||
pi.warehouse = old_packed_items_map.get((packing_item_code, main_item_row.item_code))[0].warehouse
|
|
||||||
|
|
||||||
def make_packing_list(doc):
|
def make_packing_list(doc):
|
||||||
"""make packing list for Product Bundle item"""
|
"Make/Update packing list for Product Bundle Item."
|
||||||
if doc.get("_action") and doc._action == "update_after_submit": return
|
if doc.get("_action") and doc._action == "update_after_submit":
|
||||||
|
|
||||||
parent_items = []
|
|
||||||
for d in doc.get("items"):
|
|
||||||
if frappe.db.get_value("Product Bundle", {"new_item_code": d.item_code}):
|
|
||||||
for i in get_product_bundle_items(d.item_code):
|
|
||||||
update_packing_list_item(doc, i.item_code, flt(i.qty)*flt(d.stock_qty), d, i.description)
|
|
||||||
|
|
||||||
if [d.item_code, d.name] not in parent_items:
|
|
||||||
parent_items.append([d.item_code, d.name])
|
|
||||||
|
|
||||||
cleanup_packing_list(doc, parent_items)
|
|
||||||
|
|
||||||
if frappe.db.get_single_value("Selling Settings", "editable_bundle_item_rates"):
|
|
||||||
update_product_bundle_price(doc, parent_items)
|
|
||||||
|
|
||||||
def cleanup_packing_list(doc, parent_items):
|
|
||||||
"""Remove all those child items which are no longer present in main item table"""
|
|
||||||
delete_list = []
|
|
||||||
for d in doc.get("packed_items"):
|
|
||||||
if [d.parent_item, d.parent_detail_docname] not in parent_items:
|
|
||||||
# mark for deletion from doclist
|
|
||||||
delete_list.append(d)
|
|
||||||
|
|
||||||
if not delete_list:
|
|
||||||
return doc
|
|
||||||
|
|
||||||
packed_items = doc.get("packed_items")
|
|
||||||
doc.set("packed_items", [])
|
|
||||||
|
|
||||||
for d in packed_items:
|
|
||||||
if d not in delete_list:
|
|
||||||
add_item_to_packing_list(doc, d)
|
|
||||||
|
|
||||||
def add_item_to_packing_list(doc, packed_item):
|
|
||||||
doc.append("packed_items", {
|
|
||||||
'parent_item': packed_item.parent_item,
|
|
||||||
'item_code': packed_item.item_code,
|
|
||||||
'item_name': packed_item.item_name,
|
|
||||||
'uom': packed_item.uom,
|
|
||||||
'qty': packed_item.qty,
|
|
||||||
'rate': packed_item.rate,
|
|
||||||
'conversion_factor': packed_item.conversion_factor,
|
|
||||||
'description': packed_item.description,
|
|
||||||
'warehouse': packed_item.warehouse,
|
|
||||||
'batch_no': packed_item.batch_no,
|
|
||||||
'actual_batch_qty': packed_item.actual_batch_qty,
|
|
||||||
'serial_no': packed_item.serial_no,
|
|
||||||
'target_warehouse': packed_item.target_warehouse,
|
|
||||||
'actual_qty': packed_item.actual_qty,
|
|
||||||
'projected_qty': packed_item.projected_qty,
|
|
||||||
'incoming_rate': packed_item.incoming_rate,
|
|
||||||
'prevdoc_doctype': packed_item.prevdoc_doctype,
|
|
||||||
'parent_detail_docname': packed_item.parent_detail_docname
|
|
||||||
})
|
|
||||||
|
|
||||||
def update_product_bundle_price(doc, parent_items):
|
|
||||||
"""Updates the prices of Product Bundles based on the rates of the Items in the bundle."""
|
|
||||||
|
|
||||||
if not doc.get('items'):
|
|
||||||
return
|
return
|
||||||
|
|
||||||
parent_items_index = 0
|
parent_items_price, reset = {}, False
|
||||||
bundle_price = 0
|
set_price_from_children = frappe.db.get_single_value("Selling Settings", "editable_bundle_item_rates")
|
||||||
|
|
||||||
for bundle_item in doc.get("packed_items"):
|
stale_packed_items_table = get_indexed_packed_items_table(doc)
|
||||||
if parent_items[parent_items_index][0] == bundle_item.parent_item:
|
|
||||||
bundle_item_rate = bundle_item.rate if bundle_item.rate else 0
|
|
||||||
bundle_price += bundle_item.qty * bundle_item_rate
|
|
||||||
else:
|
|
||||||
update_parent_item_price(doc, parent_items[parent_items_index][0], bundle_price)
|
|
||||||
|
|
||||||
bundle_item_rate = bundle_item.rate if bundle_item.rate else 0
|
if not doc.is_new():
|
||||||
bundle_price = bundle_item.qty * bundle_item_rate
|
reset = reset_packing_list_if_deleted_items_exist(doc)
|
||||||
parent_items_index += 1
|
|
||||||
|
|
||||||
# for the last product bundle
|
for item_row in doc.get("items"):
|
||||||
if doc.get("packed_items"):
|
if frappe.db.exists("Product Bundle", {"new_item_code": item_row.item_code}):
|
||||||
update_parent_item_price(doc, parent_items[parent_items_index][0], bundle_price)
|
for bundle_item in get_product_bundle_items(item_row.item_code):
|
||||||
|
pi_row = add_packed_item_row(
|
||||||
|
doc=doc, packing_item=bundle_item,
|
||||||
|
main_item_row=item_row, packed_items_table=stale_packed_items_table,
|
||||||
|
reset=reset
|
||||||
|
)
|
||||||
|
item_data = get_packed_item_details(bundle_item.item_code, doc.company)
|
||||||
|
update_packed_item_basic_data(item_row, pi_row, bundle_item, item_data)
|
||||||
|
update_packed_item_stock_data(item_row, pi_row, bundle_item, item_data, doc)
|
||||||
|
update_packed_item_price_data(pi_row, item_data, doc)
|
||||||
|
update_packed_item_from_cancelled_doc(item_row, bundle_item, pi_row, doc)
|
||||||
|
|
||||||
def update_parent_item_price(doc, parent_item_code, bundle_price):
|
if set_price_from_children: # create/update bundle item wise price dict
|
||||||
parent_item_doc = doc.get('items', {'item_code': parent_item_code})[0]
|
update_product_bundle_rate(parent_items_price, pi_row)
|
||||||
|
|
||||||
current_parent_item_price = parent_item_doc.amount
|
if parent_items_price:
|
||||||
if current_parent_item_price != bundle_price:
|
set_product_bundle_rate_amount(doc, parent_items_price) # set price in bundle item
|
||||||
parent_item_doc.amount = bundle_price
|
|
||||||
update_parent_item_rate(parent_item_doc, bundle_price)
|
|
||||||
|
|
||||||
def update_parent_item_rate(parent_item_doc, bundle_price):
|
def get_indexed_packed_items_table(doc):
|
||||||
parent_item_doc.rate = bundle_price/parent_item_doc.qty
|
"""
|
||||||
|
Create dict from stale packed items table like:
|
||||||
|
{(Parent Item 1, Bundle Item 1, ae4b5678): {...}, (key): {value}}
|
||||||
|
|
||||||
@frappe.whitelist()
|
Use: to quickly retrieve/check if row existed in table instead of looping n times
|
||||||
def get_items_from_product_bundle(args):
|
"""
|
||||||
args = json.loads(args)
|
indexed_table = {}
|
||||||
items = []
|
for packed_item in doc.get("packed_items"):
|
||||||
bundled_items = get_product_bundle_items(args["item_code"])
|
key = (packed_item.parent_item, packed_item.item_code, packed_item.parent_detail_docname)
|
||||||
for item in bundled_items:
|
indexed_table[key] = packed_item
|
||||||
args.update({
|
|
||||||
"item_code": item.item_code,
|
|
||||||
"qty": flt(args["quantity"]) * flt(item.qty)
|
|
||||||
})
|
|
||||||
items.append(get_item_details(args))
|
|
||||||
|
|
||||||
return items
|
return indexed_table
|
||||||
|
|
||||||
|
def reset_packing_list_if_deleted_items_exist(doc):
|
||||||
|
doc_before_save = doc.get_doc_before_save()
|
||||||
|
reset_table = False
|
||||||
|
|
||||||
|
if doc_before_save:
|
||||||
|
# reset table if:
|
||||||
|
# 1. items were deleted
|
||||||
|
# 2. if bundle item replaced by another item (same no. of items but different items)
|
||||||
|
# we maintain list to maintain repeated item rows as well
|
||||||
|
items_before_save = [item.item_code for item in doc_before_save.get("items")]
|
||||||
|
items_after_save = [item.item_code for item in doc.get("items")]
|
||||||
|
reset_table = items_before_save != items_after_save
|
||||||
|
else:
|
||||||
|
reset_table = True # reset if via Update Items (cannot determine action)
|
||||||
|
|
||||||
|
if reset_table:
|
||||||
|
doc.set("packed_items", [])
|
||||||
|
return reset_table
|
||||||
|
|
||||||
|
def get_product_bundle_items(item_code):
|
||||||
|
product_bundle = frappe.qb.DocType("Product Bundle")
|
||||||
|
product_bundle_item = frappe.qb.DocType("Product Bundle Item")
|
||||||
|
|
||||||
|
query = (
|
||||||
|
frappe.qb.from_(product_bundle_item)
|
||||||
|
.join(product_bundle).on(product_bundle_item.parent == product_bundle.name)
|
||||||
|
.select(
|
||||||
|
product_bundle_item.item_code,
|
||||||
|
product_bundle_item.qty,
|
||||||
|
product_bundle_item.uom,
|
||||||
|
product_bundle_item.description
|
||||||
|
).where(
|
||||||
|
product_bundle.new_item_code == item_code
|
||||||
|
).orderby(
|
||||||
|
product_bundle_item.idx
|
||||||
|
)
|
||||||
|
)
|
||||||
|
return query.run(as_dict=True)
|
||||||
|
|
||||||
|
def add_packed_item_row(doc, packing_item, main_item_row, packed_items_table, reset):
|
||||||
|
"""Add and return packed item row.
|
||||||
|
doc: Transaction document
|
||||||
|
packing_item (dict): Packed Item details
|
||||||
|
main_item_row (dict): Items table row corresponding to packed item
|
||||||
|
packed_items_table (dict): Packed Items table before save (indexed)
|
||||||
|
reset (bool): State if table is reset or preserved as is
|
||||||
|
"""
|
||||||
|
exists, pi_row = False, {}
|
||||||
|
|
||||||
|
# check if row already exists in packed items table
|
||||||
|
key = (main_item_row.item_code, packing_item.item_code, main_item_row.name)
|
||||||
|
if packed_items_table.get(key):
|
||||||
|
pi_row, exists = packed_items_table.get(key), True
|
||||||
|
|
||||||
|
if not exists:
|
||||||
|
pi_row = doc.append('packed_items', {})
|
||||||
|
elif reset: # add row if row exists but table is reset
|
||||||
|
pi_row.idx, pi_row.name = None, None
|
||||||
|
pi_row = doc.append('packed_items', pi_row)
|
||||||
|
|
||||||
|
return pi_row
|
||||||
|
|
||||||
|
def get_packed_item_details(item_code, company):
|
||||||
|
item = frappe.qb.DocType("Item")
|
||||||
|
item_default = frappe.qb.DocType("Item Default")
|
||||||
|
query = (
|
||||||
|
frappe.qb.from_(item)
|
||||||
|
.left_join(item_default)
|
||||||
|
.on(
|
||||||
|
(item_default.parent == item.name)
|
||||||
|
& (item_default.company == company)
|
||||||
|
).select(
|
||||||
|
item.item_name, item.is_stock_item,
|
||||||
|
item.description, item.stock_uom,
|
||||||
|
item.valuation_rate,
|
||||||
|
item_default.default_warehouse
|
||||||
|
).where(
|
||||||
|
item.name == item_code
|
||||||
|
)
|
||||||
|
)
|
||||||
|
return query.run(as_dict=True)[0]
|
||||||
|
|
||||||
|
def update_packed_item_basic_data(main_item_row, pi_row, packing_item, item_data):
|
||||||
|
pi_row.parent_item = main_item_row.item_code
|
||||||
|
pi_row.parent_detail_docname = main_item_row.name
|
||||||
|
pi_row.item_code = packing_item.item_code
|
||||||
|
pi_row.item_name = item_data.item_name
|
||||||
|
pi_row.uom = item_data.stock_uom
|
||||||
|
pi_row.qty = flt(packing_item.qty) * flt(main_item_row.stock_qty)
|
||||||
|
pi_row.conversion_factor = main_item_row.conversion_factor
|
||||||
|
|
||||||
|
if not pi_row.description:
|
||||||
|
pi_row.description = packing_item.get("description")
|
||||||
|
|
||||||
|
def update_packed_item_stock_data(main_item_row, pi_row, packing_item, item_data, doc):
|
||||||
|
# TODO batch_no, actual_batch_qty, incoming_rate
|
||||||
|
if not pi_row.warehouse and not doc.amended_from:
|
||||||
|
fetch_warehouse = (doc.get('is_pos') or item_data.is_stock_item or not item_data.default_warehouse)
|
||||||
|
pi_row.warehouse = (main_item_row.warehouse if (fetch_warehouse and main_item_row.warehouse)
|
||||||
|
else item_data.default_warehouse)
|
||||||
|
|
||||||
|
if not pi_row.target_warehouse:
|
||||||
|
pi_row.target_warehouse = main_item_row.get("target_warehouse")
|
||||||
|
|
||||||
|
bin = get_packed_item_bin_qty(packing_item.item_code, pi_row.warehouse)
|
||||||
|
pi_row.actual_qty = flt(bin.get("actual_qty"))
|
||||||
|
pi_row.projected_qty = flt(bin.get("projected_qty"))
|
||||||
|
|
||||||
|
def update_packed_item_price_data(pi_row, item_data, doc):
|
||||||
|
"Set price as per price list or from the Item master."
|
||||||
|
if pi_row.rate:
|
||||||
|
return
|
||||||
|
|
||||||
|
item_doc = frappe.get_cached_doc("Item", pi_row.item_code)
|
||||||
|
row_data = pi_row.as_dict().copy()
|
||||||
|
row_data.update({
|
||||||
|
"company": doc.get("company"),
|
||||||
|
"price_list": doc.get("selling_price_list"),
|
||||||
|
"currency": doc.get("currency")
|
||||||
|
})
|
||||||
|
rate = get_price_list_rate(row_data, item_doc).get("price_list_rate")
|
||||||
|
|
||||||
|
pi_row.rate = rate or item_data.get("valuation_rate") or 0.0
|
||||||
|
|
||||||
|
def update_packed_item_from_cancelled_doc(main_item_row, packing_item, pi_row, doc):
|
||||||
|
"Update packed item row details from cancelled doc into amended doc."
|
||||||
|
prev_doc_packed_items_map = None
|
||||||
|
if doc.amended_from:
|
||||||
|
prev_doc_packed_items_map = get_cancelled_doc_packed_item_details(doc.packed_items)
|
||||||
|
|
||||||
|
if prev_doc_packed_items_map and prev_doc_packed_items_map.get((packing_item.item_code, main_item_row.item_code)):
|
||||||
|
prev_doc_row = prev_doc_packed_items_map.get((packing_item.item_code, main_item_row.item_code))
|
||||||
|
pi_row.batch_no = prev_doc_row[0].batch_no
|
||||||
|
pi_row.serial_no = prev_doc_row[0].serial_no
|
||||||
|
pi_row.warehouse = prev_doc_row[0].warehouse
|
||||||
|
|
||||||
|
def get_packed_item_bin_qty(item, warehouse):
|
||||||
|
bin_data = frappe.db.get_values(
|
||||||
|
"Bin",
|
||||||
|
fieldname=["actual_qty", "projected_qty"],
|
||||||
|
filters={"item_code": item, "warehouse": warehouse},
|
||||||
|
as_dict=True
|
||||||
|
)
|
||||||
|
|
||||||
|
return bin_data[0] if bin_data else {}
|
||||||
|
|
||||||
|
def get_cancelled_doc_packed_item_details(old_packed_items):
|
||||||
|
prev_doc_packed_items_map = {}
|
||||||
|
for items in old_packed_items:
|
||||||
|
prev_doc_packed_items_map.setdefault((items.item_code ,items.parent_item), []).append(items.as_dict())
|
||||||
|
return prev_doc_packed_items_map
|
||||||
|
|
||||||
|
def update_product_bundle_rate(parent_items_price, pi_row):
|
||||||
|
"""
|
||||||
|
Update the price dict of Product Bundles based on the rates of the Items in the bundle.
|
||||||
|
|
||||||
|
Stucture:
|
||||||
|
{(Bundle Item 1, ae56fgji): 150.0, (Bundle Item 2, bc78fkjo): 200.0}
|
||||||
|
"""
|
||||||
|
key = (pi_row.parent_item, pi_row.parent_detail_docname)
|
||||||
|
rate = parent_items_price.get(key)
|
||||||
|
if not rate:
|
||||||
|
parent_items_price[key] = 0.0
|
||||||
|
|
||||||
|
parent_items_price[key] += flt(pi_row.rate)
|
||||||
|
|
||||||
|
def set_product_bundle_rate_amount(doc, parent_items_price):
|
||||||
|
"Set cumulative rate and amount in bundle item."
|
||||||
|
for item in doc.get("items"):
|
||||||
|
bundle_rate = parent_items_price.get((item.item_code, item.name))
|
||||||
|
if bundle_rate and bundle_rate != item.rate:
|
||||||
|
item.rate = bundle_rate
|
||||||
|
item.amount = flt(bundle_rate * item.qty)
|
||||||
|
|
||||||
def on_doctype_update():
|
def on_doctype_update():
|
||||||
frappe.db.add_index("Packed Item", ["item_code", "warehouse"])
|
frappe.db.add_index("Packed Item", ["item_code", "warehouse"])
|
||||||
|
|
||||||
def get_old_packed_item_details(old_packed_items):
|
|
||||||
old_packed_items_map = {}
|
@frappe.whitelist()
|
||||||
for items in old_packed_items:
|
def get_items_from_product_bundle(row):
|
||||||
old_packed_items_map.setdefault((items.item_code ,items.parent_item), []).append(items.as_dict())
|
row, items = json.loads(row), []
|
||||||
return old_packed_items_map
|
|
||||||
|
bundled_items = get_product_bundle_items(row["item_code"])
|
||||||
|
for item in bundled_items:
|
||||||
|
row.update({
|
||||||
|
"item_code": item.item_code,
|
||||||
|
"qty": flt(row["quantity"]) * flt(item.qty)
|
||||||
|
})
|
||||||
|
items.append(get_item_details(row))
|
||||||
|
|
||||||
|
return items
|
||||||
|
|||||||
104
erpnext/stock/doctype/packed_item/test_packed_item.py
Normal file
104
erpnext/stock/doctype/packed_item/test_packed_item.py
Normal file
@ -0,0 +1,104 @@
|
|||||||
|
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
|
||||||
|
# License: GNU General Public License v3. See license.txt
|
||||||
|
|
||||||
|
from erpnext.selling.doctype.product_bundle.test_product_bundle import make_product_bundle
|
||||||
|
from erpnext.selling.doctype.sales_order.test_sales_order import make_sales_order
|
||||||
|
from erpnext.stock.doctype.item.test_item import make_item
|
||||||
|
from erpnext.tests.utils import ERPNextTestCase, change_settings
|
||||||
|
|
||||||
|
|
||||||
|
class TestPackedItem(ERPNextTestCase):
|
||||||
|
"Test impact on Packed Items table in various scenarios."
|
||||||
|
@classmethod
|
||||||
|
def setUpClass(cls) -> None:
|
||||||
|
make_item("_Test Product Bundle X", {"is_stock_item": 0})
|
||||||
|
make_item("_Test Bundle Item 1", {"is_stock_item": 1})
|
||||||
|
make_item("_Test Bundle Item 2", {"is_stock_item": 1})
|
||||||
|
make_item("_Test Normal Stock Item", {"is_stock_item": 1})
|
||||||
|
|
||||||
|
make_product_bundle(
|
||||||
|
"_Test Product Bundle X",
|
||||||
|
["_Test Bundle Item 1", "_Test Bundle Item 2"],
|
||||||
|
qty=2
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_sales_order_adding_bundle_item(self):
|
||||||
|
"Test impact on packed items if bundle item row is added."
|
||||||
|
so = make_sales_order(item_code = "_Test Product Bundle X", qty=1,
|
||||||
|
do_not_submit=True)
|
||||||
|
|
||||||
|
self.assertEqual(so.items[0].qty, 1)
|
||||||
|
self.assertEqual(len(so.packed_items), 2)
|
||||||
|
self.assertEqual(so.packed_items[0].item_code, "_Test Bundle Item 1")
|
||||||
|
self.assertEqual(so.packed_items[0].qty, 2)
|
||||||
|
|
||||||
|
def test_sales_order_updating_bundle_item(self):
|
||||||
|
"Test impact on packed items if bundle item row is updated."
|
||||||
|
so = make_sales_order(item_code = "_Test Product Bundle X", qty=1,
|
||||||
|
do_not_submit=True)
|
||||||
|
|
||||||
|
so.items[0].qty = 2 # change qty
|
||||||
|
so.save()
|
||||||
|
|
||||||
|
self.assertEqual(so.packed_items[0].qty, 4)
|
||||||
|
self.assertEqual(so.packed_items[1].qty, 4)
|
||||||
|
|
||||||
|
# change item code to non bundle item
|
||||||
|
so.items[0].item_code = "_Test Normal Stock Item"
|
||||||
|
so.save()
|
||||||
|
|
||||||
|
self.assertEqual(len(so.packed_items), 0)
|
||||||
|
|
||||||
|
def test_sales_order_recurring_bundle_item(self):
|
||||||
|
"Test impact on packed items if same bundle item is added and removed."
|
||||||
|
so_items = []
|
||||||
|
for qty in [2, 4, 6, 8]:
|
||||||
|
so_items.append({
|
||||||
|
"item_code": "_Test Product Bundle X",
|
||||||
|
"qty": qty,
|
||||||
|
"rate": 400,
|
||||||
|
"warehouse": "_Test Warehouse - _TC"
|
||||||
|
})
|
||||||
|
|
||||||
|
# create SO with recurring bundle item
|
||||||
|
so = make_sales_order(item_list=so_items, do_not_submit=True)
|
||||||
|
|
||||||
|
# check alternate rows for qty
|
||||||
|
self.assertEqual(len(so.packed_items), 8)
|
||||||
|
self.assertEqual(so.packed_items[1].item_code, "_Test Bundle Item 2")
|
||||||
|
self.assertEqual(so.packed_items[1].qty, 4)
|
||||||
|
self.assertEqual(so.packed_items[3].qty, 8)
|
||||||
|
self.assertEqual(so.packed_items[5].qty, 12)
|
||||||
|
self.assertEqual(so.packed_items[7].qty, 16)
|
||||||
|
|
||||||
|
# delete intermediate row (2nd)
|
||||||
|
del so.items[1]
|
||||||
|
so.save()
|
||||||
|
|
||||||
|
# check alternate rows for qty
|
||||||
|
self.assertEqual(len(so.packed_items), 6)
|
||||||
|
self.assertEqual(so.packed_items[1].qty, 4)
|
||||||
|
self.assertEqual(so.packed_items[3].qty, 12)
|
||||||
|
self.assertEqual(so.packed_items[5].qty, 16)
|
||||||
|
|
||||||
|
# delete last row
|
||||||
|
del so.items[2]
|
||||||
|
so.save()
|
||||||
|
|
||||||
|
# check alternate rows for qty
|
||||||
|
self.assertEqual(len(so.packed_items), 4)
|
||||||
|
self.assertEqual(so.packed_items[1].qty, 4)
|
||||||
|
self.assertEqual(so.packed_items[3].qty, 12)
|
||||||
|
|
||||||
|
@change_settings("Selling Settings", {"editable_bundle_item_rates": 1})
|
||||||
|
def test_sales_order_bundle_item_cumulative_price(self):
|
||||||
|
"Test if Bundle Item rate is cumulative from packed items."
|
||||||
|
so = make_sales_order(item_code = "_Test Product Bundle X", qty=2,
|
||||||
|
do_not_submit=True)
|
||||||
|
|
||||||
|
so.packed_items[0].rate = 150
|
||||||
|
so.packed_items[1].rate = 200
|
||||||
|
so.save()
|
||||||
|
|
||||||
|
self.assertEqual(so.items[0].rate, 350)
|
||||||
|
self.assertEqual(so.items[0].amount, 700)
|
||||||
Loading…
x
Reference in New Issue
Block a user