test: test case to check use serial / batch fields feature

This commit is contained in:
Rohit Waghchaure 2024-02-06 13:31:36 +05:30
parent c1e869f040
commit 01650120d4
10 changed files with 208 additions and 49 deletions

View File

@ -21,6 +21,9 @@ from erpnext.stock import get_warehouse_account_map
from erpnext.stock.doctype.inventory_dimension.inventory_dimension import (
get_evaluated_inventory_dimension,
)
from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
get_type_of_transaction,
)
from erpnext.stock.stock_ledger import get_items_to_be_repost
@ -150,6 +153,13 @@ class StockController(AccountsController):
if row.use_serial_batch_fields and (
not row.serial_and_batch_bundle and not row.get("rejected_serial_and_batch_bundle")
):
if self.doctype == "Stock Reconciliation":
qty = row.qty
type_of_transaction = "Inward"
else:
qty = row.stock_qty
type_of_transaction = get_type_of_transaction(self, row)
sn_doc = SerialBatchCreation(
{
"item_code": row.item_code,
@ -159,14 +169,15 @@ class StockController(AccountsController):
"voucher_type": self.doctype,
"voucher_no": self.name,
"voucher_detail_no": row.name,
"qty": row.stock_qty,
"type_of_transaction": "Inward" if row.stock_qty > 0 else "Outward",
"qty": qty,
"type_of_transaction": type_of_transaction,
"company": self.company,
"is_rejected": 1 if row.get("rejected_warehouse") else 0,
"serial_nos": get_serial_nos(row.serial_no) if row.serial_no else None,
"batches": frappe._dict({row.batch_no: row.stock_qty}) if row.batch_no else None,
"batches": frappe._dict({row.batch_no: qty}) if row.batch_no else None,
"batch_no": row.batch_no,
"use_serial_batch_fields": row.use_serial_batch_fields,
"do_not_submit": True,
}
).make_serial_and_batch_bundle()

View File

@ -1566,7 +1566,7 @@ def create_delivery_note(**args):
dn.return_against = args.return_against
bundle_id = None
if args.get("batch_no") or args.get("serial_no"):
if not args.use_serial_batch_fields and (args.get("batch_no") or args.get("serial_no")):
type_of_transaction = args.type_of_transaction or "Outward"
if dn.is_return:
@ -1608,6 +1608,9 @@ def create_delivery_note(**args):
"expense_account": args.expense_account or "Cost of Goods Sold - _TC",
"cost_center": args.cost_center or "_Test Cost Center - _TC",
"target_warehouse": args.target_warehouse,
"use_serial_batch_fields": args.use_serial_batch_fields,
"serial_no": args.serial_no if args.use_serial_batch_fields else None,
"batch_no": args.batch_no if args.use_serial_batch_fields else None,
},
)

View File

@ -2230,6 +2230,93 @@ class TestPurchaseReceipt(FrappeTestCase):
pr_doc.reload()
self.assertFalse(pr_doc.items[0].from_warehouse)
def test_use_serial_batch_fields_for_serial_nos(self):
from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
from erpnext.stock.doctype.stock_reconciliation.test_stock_reconciliation import (
create_stock_reconciliation,
)
item_code = make_item(
"_Test Use Serial Fields Item Serial Item",
properties={"has_serial_no": 1, "serial_no_series": "SNU-TSFISI-.#####"},
).name
serial_nos = [
"SNU-TSFISI-000011",
"SNU-TSFISI-000012",
"SNU-TSFISI-000013",
"SNU-TSFISI-000014",
"SNU-TSFISI-000015",
]
pr = make_purchase_receipt(
item_code=item_code,
qty=5,
serial_no="\n".join(serial_nos),
use_serial_batch_fields=1,
rate=100,
)
self.assertEqual(pr.items[0].use_serial_batch_fields, 1)
self.assertFalse(pr.items[0].serial_no)
self.assertTrue(pr.items[0].serial_and_batch_bundle)
sbb_doc = frappe.get_doc("Serial and Batch Bundle", pr.items[0].serial_and_batch_bundle)
for row in sbb_doc.entries:
self.assertTrue(row.serial_no in serial_nos)
serial_nos.remove("SNU-TSFISI-000015")
sr = create_stock_reconciliation(
item_code=item_code,
serial_no="\n".join(serial_nos),
qty=4,
warehouse=pr.items[0].warehouse,
use_serial_batch_fields=1,
do_not_submit=True,
)
sr.reload()
serial_nos = get_serial_nos(sr.items[0].current_serial_no)
self.assertEqual(len(serial_nos), 5)
self.assertEqual(sr.items[0].current_qty, 5)
new_serial_nos = get_serial_nos(sr.items[0].serial_no)
self.assertEqual(len(new_serial_nos), 4)
self.assertEqual(sr.items[0].qty, 4)
self.assertEqual(sr.items[0].use_serial_batch_fields, 1)
self.assertFalse(sr.items[0].current_serial_and_batch_bundle)
self.assertFalse(sr.items[0].serial_and_batch_bundle)
self.assertTrue(sr.items[0].current_serial_no)
sr.submit()
sr.reload()
self.assertTrue(sr.items[0].current_serial_and_batch_bundle)
self.assertTrue(sr.items[0].serial_and_batch_bundle)
serial_no_status = frappe.db.get_value("Serial No", "SNU-TSFISI-000015", "status")
self.assertTrue(serial_no_status != "Active")
dn = create_delivery_note(
item_code=item_code,
qty=4,
serial_no="\n".join(new_serial_nos),
use_serial_batch_fields=1,
)
self.assertTrue(dn.items[0].serial_and_batch_bundle)
self.assertEqual(dn.items[0].qty, 4)
doc = frappe.get_doc("Serial and Batch Bundle", dn.items[0].serial_and_batch_bundle)
for row in doc.entries:
self.assertTrue(row.serial_no in new_serial_nos)
for sn in new_serial_nos:
serial_no_status = frappe.db.get_value("Serial No", sn, "status")
self.assertTrue(serial_no_status != "Active")
def prepare_data_for_internal_transfer():
from erpnext.accounts.doctype.sales_invoice.test_sales_invoice import create_internal_supplier
@ -2399,7 +2486,7 @@ def make_purchase_receipt(**args):
uom = args.uom or frappe.db.get_value("Item", item_code, "stock_uom") or "_Test UOM"
bundle_id = None
if args.get("batch_no") or args.get("serial_no"):
if not args.use_serial_batch_fields and (args.get("batch_no") or args.get("serial_no")):
batches = {}
if args.get("batch_no"):
batches = frappe._dict({args.batch_no: qty})
@ -2441,6 +2528,9 @@ def make_purchase_receipt(**args):
"cost_center": args.cost_center
or frappe.get_cached_value("Company", pr.company, "cost_center"),
"asset_location": args.location or "Test Location",
"use_serial_batch_fields": args.use_serial_batch_fields or 0,
"serial_no": args.serial_no if args.use_serial_batch_fields else "",
"batch_no": args.batch_no if args.use_serial_batch_fields else "",
},
)

View File

@ -1117,7 +1117,7 @@ def parse_serial_nos(data):
if isinstance(data, list):
return data
return [s.strip() for s in cstr(data).strip().upper().replace(",", "\n").split("\n") if s.strip()]
return [s.strip() for s in cstr(data).strip().replace(",", "\n").split("\n") if s.strip()]
@frappe.whitelist()
@ -1256,7 +1256,7 @@ def create_serial_batch_no_ledgers(
def get_type_of_transaction(parent_doc, child_row):
type_of_transaction = child_row.type_of_transaction
type_of_transaction = child_row.get("type_of_transaction")
if parent_doc.get("doctype") == "Stock Entry":
type_of_transaction = "Outward" if child_row.s_warehouse else "Inward"
@ -1384,6 +1384,8 @@ def get_available_serial_nos(kwargs):
filters = {"item_code": kwargs.item_code}
# ignore_warehouse is used for backdated stock transactions
# There might be chances that the serial no not exists in the warehouse during backdated stock transactions
if not kwargs.get("ignore_warehouse"):
filters["warehouse"] = ("is", "set")
if kwargs.warehouse:

View File

@ -151,9 +151,7 @@ def get_serial_nos(serial_no):
if isinstance(serial_no, list):
return serial_no
return [
s.strip() for s in cstr(serial_no).strip().upper().replace(",", "\n").split("\n") if s.strip()
]
return [s.strip() for s in cstr(serial_no).strip().replace(",", "\n").split("\n") if s.strip()]
def clean_serial_no_string(serial_no: str) -> str:

View File

@ -198,6 +198,7 @@ frappe.ui.form.on("Stock Reconciliation", {
frappe.model.set_value(cdt, cdn, "current_amount", r.message.rate * r.message.qty);
frappe.model.set_value(cdt, cdn, "amount", row.qty * row.valuation_rate);
frappe.model.set_value(cdt, cdn, "current_serial_no", r.message.serial_nos);
frappe.model.set_value(cdt, cdn, "use_serial_batch_fields", r.message.use_serial_batch_fields);
if (frm.doc.purpose == "Stock Reconciliation" && !frm.doc.scan_mode) {
frappe.model.set_value(cdt, cdn, "serial_no", r.message.serial_nos);

View File

@ -99,6 +99,7 @@ class StockReconciliation(StockController):
)
def on_submit(self):
self.make_bundle_for_current_qty()
self.make_bundle_using_old_serial_batch_fields()
self.update_stock_ledger()
self.make_gl_entries()
@ -117,9 +118,52 @@ class StockReconciliation(StockController):
self.repost_future_sle_and_gle()
self.delete_auto_created_batches()
def make_bundle_for_current_qty(self):
from erpnext.stock.serial_batch_bundle import SerialBatchCreation
for row in self.items:
if not row.use_serial_batch_fields:
continue
if row.current_serial_and_batch_bundle:
continue
if row.current_qty and (row.current_serial_no or row.batch_no):
sn_doc = SerialBatchCreation(
{
"item_code": row.item_code,
"warehouse": row.warehouse,
"posting_date": self.posting_date,
"posting_time": self.posting_time,
"voucher_type": self.doctype,
"voucher_no": self.name,
"voucher_detail_no": row.name,
"qty": row.qty,
"type_of_transaction": "Outward",
"company": self.company,
"is_rejected": 0,
"serial_nos": get_serial_nos(row.current_serial_no) if row.current_serial_no else None,
"batches": frappe._dict({row.batch_no: row.qty}) if row.batch_no else None,
"batch_no": row.batch_no,
"do_not_submit": True,
}
).make_serial_and_batch_bundle()
row.current_serial_and_batch_bundle = sn_doc.name
row.db_set(
{
"current_serial_and_batch_bundle": sn_doc.name,
"current_serial_no": "",
"batch_no": "",
}
)
def set_current_serial_and_batch_bundle(self, voucher_detail_no=None, save=False) -> None:
"""Set Serial and Batch Bundle for each item"""
for item in self.items:
if not save and item.use_serial_batch_fields:
continue
if voucher_detail_no and voucher_detail_no != item.name:
continue
@ -230,6 +274,9 @@ class StockReconciliation(StockController):
def set_new_serial_and_batch_bundle(self):
for item in self.items:
if item.use_serial_batch_fields:
continue
if not item.qty:
continue
@ -292,8 +339,10 @@ class StockReconciliation(StockController):
inventory_dimensions_dict=inventory_dimensions_dict,
)
if (item.qty is None or item.qty == item_dict.get("qty")) and (
item.valuation_rate is None or item.valuation_rate == item_dict.get("rate")
if (
(item.qty is None or item.qty == item_dict.get("qty"))
and (item.valuation_rate is None or item.valuation_rate == item_dict.get("rate"))
and (not item.serial_no or (item.serial_no == item_dict.get("serial_nos")))
):
return False
else:
@ -304,6 +353,11 @@ class StockReconciliation(StockController):
if item.valuation_rate is None:
item.valuation_rate = item_dict.get("rate")
if item_dict.get("serial_nos"):
item.current_serial_no = item_dict.get("serial_nos")
if self.purpose == "Stock Reconciliation" and not item.serial_no and item.qty:
item.serial_no = item.current_serial_no
item.current_qty = item_dict.get("qty")
item.current_valuation_rate = item_dict.get("rate")
self.calculate_difference_amount(item, item_dict)
@ -1136,9 +1190,16 @@ def get_stock_balance_for(
has_serial_no = bool(item_dict.get("has_serial_no"))
has_batch_no = bool(item_dict.get("has_batch_no"))
use_serial_batch_fields = frappe.db.get_single_value("Stock Settings", "use_serial_batch_fields")
if not batch_no and has_batch_no:
# Not enough information to fetch data
return {"qty": 0, "rate": 0, "serial_nos": None}
return {
"qty": 0,
"rate": 0,
"serial_nos": None,
"use_serial_batch_fields": use_serial_batch_fields,
}
# TODO: fetch only selected batch's values
data = get_stock_balance(
@ -1161,7 +1222,12 @@ def get_stock_balance_for(
get_batch_qty(batch_no, warehouse, posting_date=posting_date, posting_time=posting_time) or 0
)
return {"qty": qty, "rate": rate, "serial_nos": serial_nos}
return {
"qty": qty,
"rate": rate,
"serial_nos": serial_nos,
"use_serial_batch_fields": use_serial_batch_fields,
}
@frappe.whitelist()

View File

@ -1094,7 +1094,7 @@ def create_stock_reconciliation(**args):
)
bundle_id = None
if args.batch_no or args.serial_no:
if not args.use_serial_batch_fields and (args.batch_no or args.serial_no):
batches = frappe._dict({})
if args.batch_no:
batches[args.batch_no] = args.qty
@ -1125,7 +1125,10 @@ def create_stock_reconciliation(**args):
"warehouse": args.warehouse or "_Test Warehouse - _TC",
"qty": args.qty,
"valuation_rate": args.rate,
"serial_no": args.serial_no if args.use_serial_batch_fields else None,
"batch_no": args.batch_no if args.use_serial_batch_fields else None,
"serial_and_batch_bundle": bundle_id,
"use_serial_batch_fields": args.use_serial_batch_fields,
},
)

View File

@ -423,7 +423,7 @@
"label": "Auto Reserve Stock for Sales Order on Purchase"
},
{
"default": "0",
"default": "1",
"fieldname": "use_serial_batch_fields",
"fieldtype": "Check",
"label": "Use Serial / Batch Fields"

View File

@ -11,6 +11,9 @@ from frappe.query_builder.functions import CombineDatetime, IfNull, Sum
from frappe.utils import cstr, flt, get_link_to_form, nowdate, nowtime
import erpnext
from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
get_available_serial_nos,
)
from erpnext.stock.doctype.warehouse.warehouse import get_child_warehouses
from erpnext.stock.serial_batch_bundle import BatchNoValuation, SerialNoValuation
from erpnext.stock.valuation import FIFOValuation, LIFOValuation
@ -125,7 +128,21 @@ def get_stock_balance(
if with_valuation_rate:
if with_serial_no:
serial_nos = get_serial_nos_data_after_transactions(args)
serial_no_details = get_available_serial_nos(
frappe._dict(
{
"item_code": item_code,
"warehouse": warehouse,
"posting_date": posting_date,
"posting_time": posting_time,
"ignore_warehouse": 1,
}
)
)
serial_nos = ""
if serial_no_details:
serial_nos = "\n".join(d.serial_no for d in serial_no_details)
return (
(last_entry.qty_after_transaction, last_entry.valuation_rate, serial_nos)
@ -140,38 +157,6 @@ def get_stock_balance(
return last_entry.qty_after_transaction if last_entry else 0.0
def get_serial_nos_data_after_transactions(args):
serial_nos = set()
args = frappe._dict(args)
sle = frappe.qb.DocType("Stock Ledger Entry")
stock_ledger_entries = (
frappe.qb.from_(sle)
.select("serial_no", "actual_qty")
.where(
(sle.item_code == args.item_code)
& (sle.warehouse == args.warehouse)
& (
CombineDatetime(sle.posting_date, sle.posting_time)
< CombineDatetime(args.posting_date, args.posting_time)
)
& (sle.is_cancelled == 0)
)
.orderby(sle.posting_date, sle.posting_time, sle.creation)
.run(as_dict=1)
)
for stock_ledger_entry in stock_ledger_entries:
changed_serial_no = get_serial_nos_data(stock_ledger_entry.serial_no)
if stock_ledger_entry.actual_qty > 0:
serial_nos.update(changed_serial_no)
else:
serial_nos.difference_update(changed_serial_no)
return "\n".join(serial_nos)
def get_serial_nos_data(serial_nos):
from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos