diff --git a/erpnext/stock/doctype/stock_ledger_entry/test_stock_ledger_entry.py b/erpnext/stock/doctype/stock_ledger_entry/test_stock_ledger_entry.py index 2a03cf4f19e..6d113ba4eb6 100644 --- a/erpnext/stock/doctype/stock_ledger_entry/test_stock_ledger_entry.py +++ b/erpnext/stock/doctype/stock_ledger_entry/test_stock_ledger_entry.py @@ -1,6 +1,8 @@ # Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors # See license.txt +import json + import frappe from frappe.core.page.permission_manager.permission_manager import reset from frappe.utils import add_days, today @@ -32,6 +34,27 @@ class TestStockLedgerEntry(ERPNextTestCase): frappe.db.sql("delete from `tabStock Ledger Entry` where item_code in (%s)" % (', '.join(['%s']*len(items))), items) frappe.db.sql("delete from `tabBin` where item_code in (%s)" % (', '.join(['%s']*len(items))), items) + + def assertSLEs(self, doc, expected_sles, sle_filters=None): + """ Compare sorted SLEs, useful for vouchers that create multiple SLEs for same line""" + + filters = {"voucher_no": doc.name, "voucher_type": doc.doctype, "is_cancelled": 0} + if sle_filters: + filters.update(sle_filters) + sles = frappe.get_all("Stock Ledger Entry", fields=["*"], filters=filters, + order_by="timestamp(posting_date, posting_time), creation") + + for exp_sle, act_sle in zip(expected_sles, sles): + for k, v in exp_sle.items(): + act_value = act_sle[k] + if k == "stock_queue": + act_value = json.loads(act_value) + if act_value and act_value[0][0] == 0: + # ignore empty fifo bins + continue + + self.assertEqual(v, act_value, msg=f"{k} doesn't match \n{exp_sle}\n{act_sle}") + def test_item_cost_reposting(self): company = "_Test Company" @@ -349,324 +372,6 @@ class TestStockLedgerEntry(ERPNextTestCase): frappe.set_user("Administrator") user.remove_roles("Stock Manager") -<<<<<<< HEAD -======= - def test_batchwise_item_valuation_moving_average(self): - item, warehouses, batches = setup_item_valuation_test(valuation_method="Moving Average") - - # Incoming Entries for Stock Value check - pr_entry_list = [ - (item, warehouses[0], batches[0], 1, 100), - (item, warehouses[0], batches[1], 1, 50), - (item, warehouses[0], batches[0], 1, 150), - (item, warehouses[0], batches[1], 1, 100), - ] - prs = create_purchase_receipt_entries_for_batchwise_item_valuation_test(pr_entry_list) - sle_details = fetch_sle_details_for_doc_list(prs, ['stock_value']) - sv_list = [d['stock_value'] for d in sle_details] - expected_sv = [100, 150, 300, 400] - self.assertEqual(expected_sv, sv_list, "Incorrect 'Stock Value' values") - - # Outgoing Entries for Stock Value Difference check - dn_entry_list = [ - (item, warehouses[0], batches[1], 1, 200), - (item, warehouses[0], batches[0], 1, 200), - (item, warehouses[0], batches[1], 1, 200), - (item, warehouses[0], batches[0], 1, 200) - ] - dns = create_delivery_note_entries_for_batchwise_item_valuation_test(dn_entry_list) - sle_details = fetch_sle_details_for_doc_list(dns, ['stock_value_difference']) - svd_list = [-1 * d['stock_value_difference'] for d in sle_details] - expected_incoming_rates = expected_abs_svd = [75, 125, 75, 125] - - self.assertEqual(expected_abs_svd, svd_list, "Incorrect 'Stock Value Difference' values") - for dn, incoming_rate in zip(dns, expected_incoming_rates): - self.assertEqual( - dn.items[0].incoming_rate, incoming_rate, - "Incorrect 'Incoming Rate' values fetched for DN items" - ) - - - def assertSLEs(self, doc, expected_sles, sle_filters=None): - """ Compare sorted SLEs, useful for vouchers that create multiple SLEs for same line""" - - filters = {"voucher_no": doc.name, "voucher_type": doc.doctype, "is_cancelled": 0} - if sle_filters: - filters.update(sle_filters) - sles = frappe.get_all("Stock Ledger Entry", fields=["*"], filters=filters, - order_by="timestamp(posting_date, posting_time), creation") - - for exp_sle, act_sle in zip(expected_sles, sles): - for k, v in exp_sle.items(): - act_value = act_sle[k] - if k == "stock_queue": - act_value = json.loads(act_value) - if act_value and act_value[0][0] == 0: - # ignore empty fifo bins - continue - - self.assertEqual(v, act_value, msg=f"{k} doesn't match \n{exp_sle}\n{act_sle}") - - - def test_batchwise_item_valuation_stock_reco(self): - item, warehouses, batches = setup_item_valuation_test() - state = { - "stock_value" : 0.0, - "qty": 0.0 - } - def update_invariants(exp_sles): - for sle in exp_sles: - state["stock_value"] += sle["stock_value_difference"] - state["qty"] += sle["actual_qty"] - sle["stock_value"] = state["stock_value"] - sle["qty_after_transaction"] = state["qty"] - - osr1 = create_stock_reconciliation(warehouse=warehouses[0], item_code=item, qty=10, rate=100, batch_no=batches[1]) - expected_sles = [ - {"actual_qty": 10, "stock_value_difference": 1000}, - ] - update_invariants(expected_sles) - self.assertSLEs(osr1, expected_sles) - - osr2 = create_stock_reconciliation(warehouse=warehouses[0], item_code=item, qty=13, rate=200, batch_no=batches[0]) - expected_sles = [ - {"actual_qty": 13, "stock_value_difference": 200*13}, - ] - update_invariants(expected_sles) - self.assertSLEs(osr2, expected_sles) - - sr1 = create_stock_reconciliation(warehouse=warehouses[0], item_code=item, qty=5, rate=50, batch_no=batches[1]) - - expected_sles = [ - {"actual_qty": -10, "stock_value_difference": -10 * 100}, - {"actual_qty": 5, "stock_value_difference": 250} - ] - update_invariants(expected_sles) - self.assertSLEs(sr1, expected_sles) - - sr2 = create_stock_reconciliation(warehouse=warehouses[0], item_code=item, qty=20, rate=75, batch_no=batches[0]) - expected_sles = [ - {"actual_qty": -13, "stock_value_difference": -13 * 200}, - {"actual_qty": 20, "stock_value_difference": 20 * 75} - ] - update_invariants(expected_sles) - self.assertSLEs(sr2, expected_sles) - - def test_batch_wise_valuation_across_warehouse(self): - item_code, warehouses, batches = setup_item_valuation_test() - source = warehouses[0] - target = warehouses[1] - - unrelated_batch = make_stock_entry(item_code=item_code, target=source, batch_no=batches[1], - qty=5, rate=10) - self.assertSLEs(unrelated_batch, [ - {"actual_qty": 5, "stock_value_difference": 10 * 5}, - ]) - - reciept = make_stock_entry(item_code=item_code, target=source, batch_no=batches[0], qty=5, rate=10) - self.assertSLEs(reciept, [ - {"actual_qty": 5, "stock_value_difference": 10 * 5}, - ]) - - transfer = make_stock_entry(item_code=item_code, source=source, target=target, batch_no=batches[0], qty=5) - self.assertSLEs(transfer, [ - {"actual_qty": -5, "stock_value_difference": -10 * 5, "warehouse": source}, - {"actual_qty": 5, "stock_value_difference": 10 * 5, "warehouse": target} - ]) - - backdated_receipt = make_stock_entry(item_code=item_code, target=source, batch_no=batches[0], - qty=5, rate=20, posting_date=add_days(today(), -1)) - self.assertSLEs(backdated_receipt, [ - {"actual_qty": 5, "stock_value_difference": 20 * 5}, - ]) - - # check reposted average rate in *future* transfer - self.assertSLEs(transfer, [ - {"actual_qty": -5, "stock_value_difference": -15 * 5, "warehouse": source, "stock_value": 15 * 5 + 10 * 5}, - {"actual_qty": 5, "stock_value_difference": 15 * 5, "warehouse": target, "stock_value": 15 * 5} - ]) - - transfer_unrelated = make_stock_entry(item_code=item_code, source=source, - target=target, batch_no=batches[1], qty=5) - self.assertSLEs(transfer_unrelated, [ - {"actual_qty": -5, "stock_value_difference": -10 * 5, "warehouse": source, "stock_value": 15 * 5}, - {"actual_qty": 5, "stock_value_difference": 10 * 5, "warehouse": target, "stock_value": 15 * 5 + 10 * 5} - ]) - - def test_intermediate_average_batch_wise_valuation(self): - """ A batch has moving average up until posting time, - check if same is respected when backdated entry is inserted in middle""" - item_code, warehouses, batches = setup_item_valuation_test() - warehouse = warehouses[0] - - batch = batches[0] - - yesterday = make_stock_entry(item_code=item_code, target=warehouse, batch_no=batch, - qty=1, rate=10, posting_date=add_days(today(), -1)) - self.assertSLEs(yesterday, [ - {"actual_qty": 1, "stock_value_difference": 10}, - ]) - - tomorrow = make_stock_entry(item_code=item_code, target=warehouse, batch_no=batches[0], - qty=1, rate=30, posting_date=add_days(today(), 1)) - self.assertSLEs(tomorrow, [ - {"actual_qty": 1, "stock_value_difference": 30}, - ]) - - create_today = make_stock_entry(item_code=item_code, target=warehouse, batch_no=batches[0], - qty=1, rate=20) - self.assertSLEs(create_today, [ - {"actual_qty": 1, "stock_value_difference": 20}, - ]) - - consume_today = make_stock_entry(item_code=item_code, source=warehouse, batch_no=batches[0], - qty=1) - self.assertSLEs(consume_today, [ - {"actual_qty": -1, "stock_value_difference": -15}, - ]) - - consume_tomorrow = make_stock_entry(item_code=item_code, source=warehouse, batch_no=batches[0], - qty=2, posting_date=add_days(today(), 2)) - self.assertSLEs(consume_tomorrow, [ - {"stock_value_difference": -(30 + 15), "stock_value": 0, "qty_after_transaction": 0}, - ]) - - def test_legacy_item_valuation_stock_entry(self): - columns = [ - 'stock_value_difference', - 'stock_value', - 'actual_qty', - 'qty_after_transaction', - 'stock_queue', - ] - item, warehouses, batches = setup_item_valuation_test(use_batchwise_valuation=0) - - def check_sle_details_against_expected(sle_details, expected_sle_details, detail, columns): - for i, (sle_vals, ex_sle_vals) in enumerate(zip(sle_details, expected_sle_details)): - for col, sle_val, ex_sle_val in zip(columns, sle_vals, ex_sle_vals): - if col == 'stock_queue': - sle_val = get_stock_value_from_q(sle_val) - ex_sle_val = get_stock_value_from_q(ex_sle_val) - self.assertEqual( - sle_val, ex_sle_val, - f"Incorrect {col} value on transaction #: {i} in {detail}" - ) - - # List used to defer assertions to prevent commits cause of error skipped rollback - details_list = [] - - - # Test Material Receipt Entries - se_entry_list_mr = [ - (item, None, warehouses[0], batches[0], 1, 50, "2021-01-21"), - (item, None, warehouses[0], batches[1], 1, 100, "2021-01-23"), - ] - ses = create_stock_entry_entries_for_batchwise_item_valuation_test( - se_entry_list_mr, "Material Receipt" - ) - sle_details = fetch_sle_details_for_doc_list(ses, columns=columns, as_dict=0) - expected_sle_details = [ - (50.0, 50.0, 1.0, 1.0, '[[1.0, 50.0]]'), - (100.0, 150.0, 1.0, 2.0, '[[1.0, 50.0], [1.0, 100.0]]'), - ] - details_list.append(( - sle_details, expected_sle_details, - "Material Receipt Entries", columns - )) - - - # Test Material Issue Entries - se_entry_list_mi = [ - (item, warehouses[0], None, batches[1], 1, None, "2021-01-29"), - ] - ses = create_stock_entry_entries_for_batchwise_item_valuation_test( - se_entry_list_mi, "Material Issue" - ) - sle_details = fetch_sle_details_for_doc_list(ses, columns=columns, as_dict=0) - expected_sle_details = [ - (-50.0, 100.0, -1.0, 1.0, '[[1, 100.0]]') - ] - details_list.append(( - sle_details, expected_sle_details, - "Material Issue Entries", columns - )) - - - # Run assertions - for details in details_list: - check_sle_details_against_expected(*details) - - def test_mixed_valuation_batches_fifo(self): - item_code, warehouses, batches = setup_item_valuation_test(use_batchwise_valuation=0) - warehouse = warehouses[0] - - state = { - "qty": 0.0, - "stock_value": 0.0 - } - def update_invariants(exp_sles): - for sle in exp_sles: - state["stock_value"] += sle["stock_value_difference"] - state["qty"] += sle["actual_qty"] - sle["stock_value"] = state["stock_value"] - sle["qty_after_transaction"] = state["qty"] - return exp_sles - - old1 = make_stock_entry(item_code=item_code, target=warehouse, batch_no=batches[0], - qty=10, rate=10) - self.assertSLEs(old1, update_invariants([ - {"actual_qty": 10, "stock_value_difference": 10*10, "stock_queue": [[10, 10]]}, - ])) - old2 = make_stock_entry(item_code=item_code, target=warehouse, batch_no=batches[1], - qty=10, rate=20) - self.assertSLEs(old2, update_invariants([ - {"actual_qty": 10, "stock_value_difference": 10*20, "stock_queue": [[10, 10], [10, 20]]}, - ])) - old3 = make_stock_entry(item_code=item_code, target=warehouse, batch_no=batches[0], - qty=5, rate=15) - - self.assertSLEs(old3, update_invariants([ - {"actual_qty": 5, "stock_value_difference": 5*15, "stock_queue": [[10, 10], [10, 20], [5, 15]]}, - ])) - - new1 = make_stock_entry(item_code=item_code, target=warehouse, qty=10, rate=40) - batches.append(new1.items[0].batch_no) - # assert old queue remains - self.assertSLEs(new1, update_invariants([ - {"actual_qty": 10, "stock_value_difference": 10*40, "stock_queue": [[10, 10], [10, 20], [5, 15]]}, - ])) - - new2 = make_stock_entry(item_code=item_code, target=warehouse, qty=10, rate=42) - batches.append(new2.items[0].batch_no) - self.assertSLEs(new2, update_invariants([ - {"actual_qty": 10, "stock_value_difference": 10*42, "stock_queue": [[10, 10], [10, 20], [5, 15]]}, - ])) - - # consume old batch as per FIFO - consume_old1 = make_stock_entry(item_code=item_code, source=warehouse, qty=15, batch_no=batches[0]) - self.assertSLEs(consume_old1, update_invariants([ - {"actual_qty": -15, "stock_value_difference": -10*10 - 5*20, "stock_queue": [[5, 20], [5, 15]]}, - ])) - - # consume new batch as per batch - consume_new2 = make_stock_entry(item_code=item_code, source=warehouse, qty=10, batch_no=batches[-1]) - self.assertSLEs(consume_new2, update_invariants([ - {"actual_qty": -10, "stock_value_difference": -10*42, "stock_queue": [[5, 20], [5, 15]]}, - ])) - - # finish all old batches - consume_old2 = make_stock_entry(item_code=item_code, source=warehouse, qty=10, batch_no=batches[1]) - self.assertSLEs(consume_old2, update_invariants([ - {"actual_qty": -10, "stock_value_difference": -5*20 - 5*15, "stock_queue": []}, - ])) - - # finish all new batches - consume_new1 = make_stock_entry(item_code=item_code, source=warehouse, qty=10, batch_no=batches[-2]) - self.assertSLEs(consume_new1, update_invariants([ - {"actual_qty": -10, "stock_value_difference": -10*40, "stock_queue": []}, - ])) ->>>>>>> b3b7cdfb49 (test: FIFO transfer for multi-batch transaction) - def test_fifo_dependent_consumption(self): item = make_item("_TestFifoTransferRates") source = "_Test Warehouse - _TC"