test: test cases for serial and batch bundle

This commit is contained in:
Rohit Waghchaure 2023-06-01 16:08:49 +05:30
parent 42b229435c
commit 40ab3bdd35
6 changed files with 436 additions and 138 deletions

View File

@ -1,5 +1,3 @@
from collections import defaultdict
import frappe
from frappe.query_builder.functions import CombineDatetime, Sum
from frappe.utils import flt
@ -118,25 +116,38 @@ class DeprecatedBatchNoValuation:
if not self.non_batchwise_valuation_batches:
return
avg_rate = self.get_avg_rate_for_non_batchwise_valuation_batches()
avilable_qty = self.get_available_qty_for_non_batchwise_valuation_batches()
self.non_batchwise_balance_value = 0.0
self.non_batchwise_balance_qty = 0.0
for batch_no in self.non_batchwise_valuation_batches:
self.stock_value_differece[batch_no] = avg_rate
self.available_qty[batch_no] = avilable_qty.get(batch_no, 0)
self.set_balance_value_for_non_batchwise_valuation_batches()
for batch_no, ledger in self.batch_nos.items():
if batch_no not in self.non_batchwise_valuation_batches:
continue
self.batch_avg_rate[batch_no] = (
self.non_batchwise_balance_value / self.non_batchwise_balance_qty
)
stock_value_change = self.batch_avg_rate[batch_no] * ledger.qty
self.stock_value_change += stock_value_change
frappe.db.set_value(
"Serial and Batch Entry",
ledger.name,
{
"stock_value_difference": stock_value_change,
"incoming_rate": self.batch_avg_rate[batch_no],
},
)
@deprecated
def get_avg_rate_for_non_batchwise_valuation_batches(self):
stock_value, qty = self.get_balance_value_and_qty_from_sl_entries()
stock_value, qty = self.get_balance_value_and_qty_from_bundle(stock_value, qty)
return stock_value / qty if qty else 0
def set_balance_value_for_non_batchwise_valuation_batches(self):
self.set_balance_value_from_sl_entries()
self.set_balance_value_from_bundle()
@deprecated
def get_balance_value_and_qty_from_sl_entries(self):
stock_value_difference = 0.0
available_qty = 0.0
def set_balance_value_from_sl_entries(self) -> None:
sle = frappe.qb.DocType("Stock Ledger Entry")
batch = frappe.qb.DocType("Batch")
@ -154,8 +165,9 @@ class DeprecatedBatchNoValuation:
.inner_join(batch)
.on(sle.batch_no == batch.name)
.select(
Sum(sle.stock_value_difference).as_("batch_value"),
sle.batch_no,
Sum(sle.actual_qty).as_("batch_qty"),
Sum(sle.stock_value_difference).as_("batch_value"),
)
.where(
(sle.item_code == self.sle.item_code)
@ -165,19 +177,19 @@ class DeprecatedBatchNoValuation:
& (sle.is_cancelled == 0)
)
.where(timestamp_condition)
.groupby(sle.batch_no)
)
if self.sle.name:
query = query.where(sle.name != self.sle.name)
for d in query.run(as_dict=True):
stock_value_difference += flt(d.batch_value)
available_qty += flt(d.batch_qty)
return stock_value_difference, available_qty
self.non_batchwise_balance_value += flt(d.batch_value)
self.non_batchwise_balance_qty += flt(d.batch_qty)
self.available_qty[d.batch_no] += flt(d.batch_qty)
@deprecated
def get_balance_value_and_qty_from_bundle(self, stock_value, qty):
def set_balance_value_from_bundle(self) -> None:
bundle = frappe.qb.DocType("Serial and Batch Bundle")
bundle_child = frappe.qb.DocType("Serial and Batch Entry")
batch = frappe.qb.DocType("Batch")
@ -199,8 +211,9 @@ class DeprecatedBatchNoValuation:
.inner_join(batch)
.on(bundle_child.batch_no == batch.name)
.select(
Sum(bundle_child.stock_value_difference).as_("batch_value"),
bundle_child.batch_no,
Sum(bundle_child.qty).as_("batch_qty"),
Sum(bundle_child.stock_value_difference).as_("batch_value"),
)
.where(
(bundle.item_code == self.sle.item_code)
@ -208,93 +221,10 @@ class DeprecatedBatchNoValuation:
& (bundle_child.batch_no.isnotnull())
& (batch.use_batchwise_valuation == 0)
& (bundle.is_cancelled == 0)
& (bundle.docstatus == 1)
& (bundle.type_of_transaction.isin(["Inward", "Outward"]))
)
.where(timestamp_condition)
)
if self.sle.serial_and_batch_bundle:
query = query.where(bundle.name != self.sle.serial_and_batch_bundle)
for d in query.run(as_dict=True):
stock_value += flt(d.batch_value)
qty += flt(d.batch_qty)
return stock_value, qty
@deprecated
def get_available_qty_for_non_batchwise_valuation_batches(self):
available_qty = defaultdict(float)
self.set_available_qty_for_non_batchwise_valuation_batches_from_sle(available_qty)
self.set_available_qty_for_non_batchwise_valuation_batches_from_bundle(available_qty)
return available_qty
@deprecated
def set_available_qty_for_non_batchwise_valuation_batches_from_sle(self, available_qty):
sle = frappe.qb.DocType("Stock Ledger Entry")
timestamp_condition = CombineDatetime(sle.posting_date, sle.posting_time) < CombineDatetime(
self.sle.posting_date, self.sle.posting_time
)
if self.sle.creation:
timestamp_condition |= (
CombineDatetime(sle.posting_date, sle.posting_time)
== CombineDatetime(self.sle.posting_date, self.sle.posting_time)
) & (sle.creation < self.sle.creation)
query = (
frappe.qb.from_(sle)
.select(
sle.batch_no,
Sum(sle.actual_qty).as_("batch_qty"),
)
.where(
(sle.item_code == self.sle.item_code)
& (sle.warehouse == self.sle.warehouse)
& (sle.batch_no.isin(self.non_batchwise_valuation_batches))
& (sle.is_cancelled == 0)
)
.where(timestamp_condition)
.groupby(sle.batch_no)
)
if self.sle.name:
query = query.where(sle.name != self.sle.name)
for d in query.run(as_dict=True):
available_qty[d.batch_no] += flt(d.batch_qty)
@deprecated
def set_available_qty_for_non_batchwise_valuation_batches_from_bundle(self, available_qty):
bundle = frappe.qb.DocType("Serial and Batch Bundle")
bundle_child = frappe.qb.DocType("Serial and Batch Entry")
timestamp_condition = CombineDatetime(
bundle.posting_date, bundle.posting_time
) < CombineDatetime(self.sle.posting_date, self.sle.posting_time)
if self.sle.creation:
timestamp_condition |= (
CombineDatetime(bundle.posting_date, bundle.posting_time)
== CombineDatetime(self.sle.posting_date, self.sle.posting_time)
) & (bundle.creation < self.sle.creation)
query = (
frappe.qb.from_(bundle)
.inner_join(bundle_child)
.on(bundle.name == bundle_child.parent)
.select(
bundle_child.batch_no,
Sum(bundle_child.qty).as_("batch_qty"),
)
.where(
(bundle.item_code == self.sle.item_code)
& (bundle.warehouse == self.sle.warehouse)
& (bundle_child.batch_no.isin(self.non_batchwise_valuation_batches))
& (bundle.is_cancelled == 0)
)
.where(timestamp_condition)
.groupby(bundle_child.batch_no)
)
@ -302,4 +232,6 @@ class DeprecatedBatchNoValuation:
query = query.where(bundle.name != self.sle.serial_and_batch_bundle)
for d in query.run(as_dict=True):
available_qty[d.batch_no] += flt(d.batch_qty)
self.non_batchwise_balance_value += flt(d.batch_value)
self.non_batchwise_balance_qty += flt(d.batch_qty)
self.available_qty[d.batch_no] += flt(d.batch_qty)

View File

@ -130,9 +130,7 @@ class Batch(Document):
frappe.throw(_("The selected item cannot have Batch"))
def set_batchwise_valuation(self):
from erpnext.stock.stock_ledger import get_valuation_method
if self.is_new() and get_valuation_method(self.item) != "Moving Average":
if self.is_new():
self.use_batchwise_valuation = 1
def before_save(self):

View File

@ -179,6 +179,7 @@ class SerialandBatchBundle(Document):
self.validate_negative_batch(d.batch_no, available_qty)
d.stock_value_difference = flt(d.qty) * flt(d.incoming_rate)
if save:
d.db_set(
{"incoming_rate": d.incoming_rate, "stock_value_difference": d.stock_value_difference}
@ -458,6 +459,8 @@ class SerialandBatchBundle(Document):
serial_nos = []
batch_nos = []
serial_batches = {}
for row in self.entries:
if row.serial_no:
serial_nos.append(row.serial_no)
@ -465,12 +468,34 @@ class SerialandBatchBundle(Document):
if row.batch_no and not row.serial_no:
batch_nos.append(row.batch_no)
if row.serial_no and row.batch_no and self.type_of_transaction == "Outward":
serial_batches.setdefault(row.serial_no, row.batch_no)
if serial_nos:
self.validate_incorrect_serial_nos(serial_nos)
elif batch_nos:
self.validate_incorrect_batch_nos(batch_nos)
if serial_batches:
self.validate_serial_batch_no(serial_batches)
def validate_serial_batch_no(self, serial_batches):
correct_batches = frappe._dict(
frappe.get_all(
"Serial No",
filters={"name": ("in", list(serial_batches.keys()))},
fields=["name", "batch_no"],
as_list=True,
)
)
for serial_no, batch_no in serial_batches.items():
if correct_batches.get(serial_no) != batch_no:
self.throw_error_message(
f"Serial No {bold(serial_no)} does not belong to Batch No {bold(batch_no)}"
)
def validate_incorrect_serial_nos(self, serial_nos):
if self.voucher_type == "Stock Entry" and self.voucher_no:

View File

@ -1,57 +1,385 @@
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
# See license.txt
# import frappe
from frappe.tests.utils import FrappeTestCase
import json
from erpnext.stock.serial_batch_bundle import get_batch_nos, get_serial_nos
import frappe
from frappe.tests.utils import FrappeTestCase
from frappe.utils import add_days, add_to_date, flt, nowdate, nowtime, today
from erpnext.stock.doctype.item.test_item import make_item
from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
class TestSerialandBatchBundle(FrappeTestCase):
def test_inward_serial_batch_bundle(self):
pass
def test_inward_outward_serial_valuation(self):
from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt
def test_outward_serial_batch_bundle(self):
pass
serial_item_code = "New Serial No Valuation 1"
make_item(
serial_item_code,
{
"has_serial_no": 1,
"serial_no_series": "TEST-SER-VAL-.#####",
"is_stock_item": 1,
},
)
pr = make_purchase_receipt(
item_code=serial_item_code, warehouse="_Test Warehouse - _TC", qty=1, rate=500
)
serial_no1 = get_serial_nos_from_bundle(pr.items[0].serial_and_batch_bundle)[0]
pr = make_purchase_receipt(
item_code=serial_item_code, warehouse="_Test Warehouse - _TC", qty=1, rate=300
)
serial_no2 = get_serial_nos_from_bundle(pr.items[0].serial_and_batch_bundle)[0]
dn = create_delivery_note(
item_code=serial_item_code,
warehouse="_Test Warehouse - _TC",
qty=1,
rate=1500,
serial_no=[serial_no2],
)
stock_value_difference = frappe.db.get_value(
"Stock Ledger Entry",
{"voucher_no": dn.name, "is_cancelled": 0, "voucher_type": "Delivery Note"},
"stock_value_difference",
)
self.assertEqual(flt(stock_value_difference, 2), -300)
dn = create_delivery_note(
item_code=serial_item_code,
warehouse="_Test Warehouse - _TC",
qty=1,
rate=1500,
serial_no=[serial_no1],
)
stock_value_difference = frappe.db.get_value(
"Stock Ledger Entry",
{"voucher_no": dn.name, "is_cancelled": 0, "voucher_type": "Delivery Note"},
"stock_value_difference",
)
self.assertEqual(flt(stock_value_difference, 2), -500)
def test_inward_outward_batch_valuation(self):
from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt
batch_item_code = "New Batch No Valuation 1"
make_item(
batch_item_code,
{
"has_batch_no": 1,
"create_new_batch": 1,
"batch_number_series": "TEST-BATTCCH-VAL-.#####",
"is_stock_item": 1,
},
)
pr = make_purchase_receipt(
item_code=batch_item_code, warehouse="_Test Warehouse - _TC", qty=10, rate=500
)
batch_no1 = get_batch_from_bundle(pr.items[0].serial_and_batch_bundle)
pr = make_purchase_receipt(
item_code=batch_item_code, warehouse="_Test Warehouse - _TC", qty=10, rate=300
)
batch_no2 = get_batch_from_bundle(pr.items[0].serial_and_batch_bundle)
dn = create_delivery_note(
item_code=batch_item_code,
warehouse="_Test Warehouse - _TC",
qty=10,
rate=1500,
batch_no=batch_no2,
)
stock_value_difference = frappe.db.get_value(
"Stock Ledger Entry",
{"voucher_no": dn.name, "is_cancelled": 0, "voucher_type": "Delivery Note"},
"stock_value_difference",
)
self.assertEqual(flt(stock_value_difference, 2), -3000)
dn = create_delivery_note(
item_code=batch_item_code,
warehouse="_Test Warehouse - _TC",
qty=10,
rate=1500,
batch_no=batch_no1,
)
stock_value_difference = frappe.db.get_value(
"Stock Ledger Entry",
{"voucher_no": dn.name, "is_cancelled": 0, "voucher_type": "Delivery Note"},
"stock_value_difference",
)
self.assertEqual(flt(stock_value_difference, 2), -5000)
def test_old_batch_valuation(self):
pass
frappe.flags.ignore_serial_batch_bundle_validation = True
batch_item_code = "Old Batch Item Valuation 1"
make_item(
batch_item_code,
{
"has_batch_no": 1,
"is_stock_item": 1,
},
)
def test_old_batch_batchwise_valuation(self):
pass
batch_id = "Old Batch 1"
if not frappe.db.exists("Batch", batch_id):
batch_doc = frappe.get_doc(
{
"doctype": "Batch",
"batch_id": batch_id,
"item": batch_item_code,
"use_batchwise_valuation": 0,
}
).insert(ignore_permissions=True)
self.assertTrue(batch_doc.use_batchwise_valuation)
batch_doc.db_set("use_batchwise_valuation", 0)
stock_queue = []
qty_after_transaction = 0
balance_value = 0
for qty, valuation in {10: 100, 20: 200}.items():
stock_queue.append([qty, valuation])
qty_after_transaction += qty
balance_value += qty_after_transaction * valuation
doc = frappe.get_doc(
{
"doctype": "Stock Ledger Entry",
"posting_date": today(),
"posting_time": nowtime(),
"batch_no": batch_id,
"incoming_rate": valuation,
"qty_after_transaction": qty_after_transaction,
"stock_value_difference": valuation * qty,
"balance_value": balance_value,
"valuation_rate": balance_value / qty_after_transaction,
"actual_qty": qty,
"item_code": batch_item_code,
"warehouse": "_Test Warehouse - _TC",
"stock_queue": json.dumps(stock_queue),
}
)
doc.flags.ignore_permissions = True
doc.flags.ignore_mandatory = True
doc.flags.ignore_links = True
doc.flags.ignore_validate = True
doc.submit()
bundle_doc = make_serial_batch_bundle(
{
"item_code": batch_item_code,
"warehouse": "_Test Warehouse - _TC",
"voucher_type": "Stock Entry",
"posting_date": today(),
"posting_time": nowtime(),
"qty": -10,
"batches": frappe._dict({batch_id: 10}),
"type_of_transaction": "Outward",
"do_not_submit": True,
}
)
bundle_doc.reload()
for row in bundle_doc.entries:
self.assertEqual(flt(row.stock_value_difference, 2), -1666.67)
bundle_doc.flags.ignore_permissions = True
bundle_doc.flags.ignore_mandatory = True
bundle_doc.flags.ignore_links = True
bundle_doc.flags.ignore_validate = True
bundle_doc.submit()
bundle_doc = make_serial_batch_bundle(
{
"item_code": batch_item_code,
"warehouse": "_Test Warehouse - _TC",
"voucher_type": "Stock Entry",
"posting_date": today(),
"posting_time": nowtime(),
"qty": -20,
"batches": frappe._dict({batch_id: 20}),
"type_of_transaction": "Outward",
"do_not_submit": True,
}
)
bundle_doc.reload()
for row in bundle_doc.entries:
self.assertEqual(flt(row.stock_value_difference, 2), -3333.33)
bundle_doc.flags.ignore_permissions = True
bundle_doc.flags.ignore_mandatory = True
bundle_doc.flags.ignore_links = True
bundle_doc.flags.ignore_validate = True
bundle_doc.submit()
frappe.flags.ignore_serial_batch_bundle_validation = False
def test_old_serial_no_valuation(self):
pass
from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt
serial_no_item_code = "Old Serial No Item Valuation 1"
make_item(
serial_no_item_code,
{
"has_serial_no": 1,
"serial_no_series": "TEST-SER-VALL-.#####",
"is_stock_item": 1,
},
)
make_purchase_receipt(
item_code=serial_no_item_code, warehouse="_Test Warehouse - _TC", qty=1, rate=500
)
frappe.flags.ignore_serial_batch_bundle_validation = True
serial_no_id = "Old Serial No 1"
if not frappe.db.exists("Serial No", serial_no_id):
sn_doc = frappe.get_doc(
{
"doctype": "Serial No",
"serial_no": serial_no_id,
"item_code": serial_no_item_code,
"company": "_Test Company",
}
).insert(ignore_permissions=True)
sn_doc.db_set(
{
"warehouse": "_Test Warehouse - _TC",
"purchase_rate": 100,
}
)
doc = frappe.get_doc(
{
"doctype": "Stock Ledger Entry",
"posting_date": today(),
"posting_time": nowtime(),
"serial_no": serial_no_id,
"incoming_rate": 100,
"qty_after_transaction": 1,
"stock_value_difference": 100,
"balance_value": 100,
"valuation_rate": 100,
"actual_qty": 1,
"item_code": serial_no_item_code,
"warehouse": "_Test Warehouse - _TC",
"company": "_Test Company",
}
)
doc.flags.ignore_permissions = True
doc.flags.ignore_mandatory = True
doc.flags.ignore_links = True
doc.flags.ignore_validate = True
doc.submit()
bundle_doc = make_serial_batch_bundle(
{
"item_code": serial_no_item_code,
"warehouse": "_Test Warehouse - _TC",
"voucher_type": "Stock Entry",
"posting_date": today(),
"posting_time": nowtime(),
"qty": -1,
"serial_nos": [serial_no_id],
"type_of_transaction": "Outward",
"do_not_submit": True,
}
)
bundle_doc.reload()
for row in bundle_doc.entries:
self.assertEqual(flt(row.stock_value_difference, 2), -100.00)
def test_batch_not_belong_to_serial_no(self):
pass
from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt
def test_serial_no_not_exists(self):
pass
serial_and_batch_code = "New Serial No Valuation 1"
make_item(
serial_and_batch_code,
{
"has_serial_no": 1,
"serial_no_series": "TEST-SER-VALL-.#####",
"is_stock_item": 1,
"has_batch_no": 1,
"create_new_batch": 1,
"batch_number_series": "TEST-SNBAT-VAL-.#####",
},
)
def test_serial_no_item(self):
pass
pr = make_purchase_receipt(
item_code=serial_and_batch_code, warehouse="_Test Warehouse - _TC", qty=1, rate=500
)
def test_serial_no_not_required(self):
pass
serial_no = get_serial_nos_from_bundle(pr.items[0].serial_and_batch_bundle)[0]
def test_serial_no_required(self):
pass
pr = make_purchase_receipt(
item_code=serial_and_batch_code, warehouse="_Test Warehouse - _TC", qty=1, rate=300
)
def test_batch_no_not_required(self):
pass
batch_no = get_batch_from_bundle(pr.items[0].serial_and_batch_bundle)
def test_batch_no_required(self):
pass
doc = frappe.get_doc(
{
"doctype": "Serial and Batch Bundle",
"item_code": serial_and_batch_code,
"warehouse": "_Test Warehouse - _TC",
"voucher_type": "Stock Entry",
"posting_date": today(),
"posting_time": nowtime(),
"qty": -1,
"type_of_transaction": "Outward",
}
)
doc.append(
"entries",
{
"batch_no": batch_no,
"serial_no": serial_no,
"qty": -1,
},
)
# Batch does not belong to serial no
self.assertRaises(frappe.exceptions.ValidationError, doc.save)
def get_batch_from_bundle(bundle):
from erpnext.stock.serial_batch_bundle import get_batch_nos
batches = get_batch_nos(bundle)
return list(batches.keys())[0]
def get_serial_nos_from_bundle(bundle):
from erpnext.stock.serial_batch_bundle import get_serial_nos
serial_nos = get_serial_nos(bundle)
return sorted(serial_nos) if serial_nos else []
@ -59,6 +387,9 @@ def get_serial_nos_from_bundle(bundle):
def make_serial_batch_bundle(kwargs):
from erpnext.stock.serial_batch_bundle import SerialBatchCreation
if isinstance(kwargs, dict):
kwargs = frappe._dict(kwargs)
type_of_transaction = "Inward" if kwargs.qty > 0 else "Outward"
if kwargs.get("type_of_transaction"):
type_of_transaction = kwargs.get("type_of_transaction")

View File

@ -52,6 +52,9 @@ class StockLedgerEntry(Document):
def on_submit(self):
self.check_stock_frozen_date()
if frappe.flags.in_test and frappe.flags.ignore_serial_batch_bundle_validation:
return
if not self.get("via_landed_cost_voucher"):
SerialBatchBundle(
sle=self,

View File

@ -434,6 +434,7 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
)
else:
entries = self.get_batch_no_ledgers()
self.stock_value_change = 0.0
self.batch_avg_rate = defaultdict(float)
self.available_qty = defaultdict(float)
self.stock_value_differece = defaultdict(float)
@ -443,6 +444,7 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
self.available_qty[ledger.batch_no] += flt(ledger.qty)
self.calculate_avg_rate_from_deprecarated_ledgers()
self.calculate_avg_rate_for_non_batchwise_valuation()
self.set_stock_value_difference()
def get_batch_no_ledgers(self) -> List[dict]:
@ -513,8 +515,10 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
return get_batch_nos(self.sle.serial_and_batch_bundle)
def set_stock_value_difference(self):
self.stock_value_change = 0
for batch_no, ledger in self.batch_nos.items():
if batch_no in self.non_batchwise_valuation_batches:
continue
if not self.available_qty[batch_no]:
continue
@ -525,8 +529,14 @@ class BatchNoValuation(DeprecatedBatchNoValuation):
# New Stock Value Difference
stock_value_change = self.batch_avg_rate[batch_no] * ledger.qty
self.stock_value_change += stock_value_change
frappe.db.set_value(
"Serial and Batch Entry", ledger.name, "stock_value_difference", stock_value_change
"Serial and Batch Entry",
ledger.name,
{
"stock_value_difference": stock_value_change,
"incoming_rate": self.batch_avg_rate[batch_no],
},
)
def calculate_valuation_rate(self):
@ -740,7 +750,6 @@ class SerialBatchCreation:
if len(batches) == 1:
self.batch_no = batches[0]
self.serial_nos = self.get_auto_created_serial_nos()
print(self.serial_nos)
def update_serial_and_batch_entries(self):
doc = frappe.get_doc("Serial and Batch Bundle", self.serial_and_batch_bundle)