Shiloh/brotherton-erpnext
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

stock reco testcases

Browse Source
This commit is contained in:
Nabin Hait 2013-01-10 10:40:37 +05:30
parent 3e28f8f54b
commit 9514d170c2
3 changed files with 100 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
stock/doctype/stock_reconciliation/stock_reconciliation.py
View File

@ -20,12 +20,14 @@ import json
from webnotes import msgprint, _
from webnotes.utils import cstr, flt
from webnotes.model.controller import DocListController
from stock.stock_ledger import update_entries_after
class DocType(DocListController):
def validate(self):
self.validate_data()
def on_submit(self):
print "in stock reco"
self.insert_stock_ledger_entries()
def on_cancel(self):
@ -110,7 +112,6 @@ class DocType(DocListController):
data = json.loads(self.doc.reconciliation_json)
for row_num, row in enumerate(data[1:]):
row = webnotes._dict(zip(row_template, row))
previous_sle = get_previous_sle({
"item_code": row.item_code,
"warehouse": row.warehouse,
@ -118,13 +119,20 @@ class DocType(DocListController):
"posting_time": self.doc.posting_time
})
change_in_qty = row.qty != "" and \
(flt(row.qty) != flt(previous_sle.get("qty_after_transaction")))
change_in_rate = row.valuation_rate != "" and \
(flt(row.valuation_rate) != flt(previous_sle.get("valuation_rate")))
if get_valuation_method(row.item_code) == "Moving Average":
self.sle_for_moving_avg(row, previous_sle)
self.sle_for_moving_avg(row, previous_sle, change_in_qty, change_in_rate)
else:
self.sle_for_fifo(row, previous_sle)
self.sle_for_fifo(row, previous_sle, change_in_qty, change_in_rate)
def sle_for_moving_avg(self, row, previous_sle):
def sle_for_moving_avg(self, row, previous_sle, change_in_qty, change_in_rate):
"""Insert Stock Ledger Entries for Moving Average valuation"""
def _get_incoming_rate(qty, valuation_rate, previous_qty, previous_valuation_rate):
if previous_valuation_rate == 0:
@ -132,12 +140,6 @@ class DocType(DocListController):
else:
return (qty * valuation_rate - previous_qty * previous_valuation_rate) \
/ flt(qty - previous_qty)
change_in_qty = row.qty != "" and \
(flt(row.qty) != flt(previous_sle.get("qty_after_transaction")))
change_in_rate = row.valuation_rate != "" and \
(flt(row.valuation_rate) != flt(previous_sle.get("valuation_rate")))
if change_in_qty:
incoming_rate = _get_incoming_rate(flt(row.qty), flt(row.valuation_rate),
@ -158,23 +160,27 @@ class DocType(DocListController):
# -1 entry
self.insert_entries({"actual_qty": -1}, row)
def sle_for_fifo(self, row, previous_sle):
def sle_for_fifo(self, row, previous_sle, change_in_qty, change_in_rate):
"""Insert Stock Ledger Entries for FIFO valuation"""
previous_stock_queue = json.loads(previous_sle.stock_queue)
previous_stock_queue = json.loads(previous_sle.stock_queue or "[]")
if previous_stock_queue != [[row.qty, row.valuation_rate]]:
# make entry as per attachment
self.insert_entries({"actual_qty": row.qty, "incoming_rate": row.valuation_rate},
row)
if change_in_qty:
if previous_stock_queue != [[row.qty, row.valuation_rate]]:
# make entry as per attachment
self.insert_entries({"actual_qty": row.qty, "incoming_rate": row.valuation_rate}, row)
# Make reverse entry
qty = sum((flt(fifo_item[0]) for fifo_item in previous_stock_queue))
self.insert_entries({"actual_qty": -1 * qty}, row)
# Make reverse entry
qty = sum((flt(fifo_item[0]) for fifo_item in previous_stock_queue))
self.insert_entries({"actual_qty": -1 * qty,
"incoming_rate": qty < 0 and row.valuation_rate or 0}, row)
elif change_in_rate:
pass
def insert_entries(self, opts, row):
"""Insert Stock Ledger Entries"""
args = {
"doctype": "Stock Ledger Entry",
"item_code": row.item_code,
"warehouse": row.warehouse,
"posting_date": self.doc.posting_date,
@ -185,8 +191,12 @@ class DocType(DocListController):
"is_cancelled": "No"
}
args.update(opts)
print args
sle_wrapper = webnotes.model_wrapper([args]).insert()
update_entries_after(args)
return webnotes.model_wrapper([args]).insert()
return sle_wrapper
def delete_stock_ledger_entries(self):
""" Delete Stock Ledger Entries related to this Stock Reconciliation

78
stock/doctype/stock_reconciliation/test_stock_reconciliation.py
View File

@ -31,23 +31,45 @@ class TestStockReconciliation(unittest.TestCase):
self.insert_test_data()
def tearDown(self):
print "Message Log:", webnotes.message_log
# print "Message Log:", "\n--\n".join(webnotes.message_log)
# print "Debug Log:", "\n--\n".join(webnotes.debug_log)
webnotes.conn.rollback()
def test_reco_for_fifo(self):
webnotes.conn.set_value("Item", "Android Jack D", "valuation_method", "FIFO")
self.submit_stock_reconciliation("2012-12-26", "12:05", 50, 1000)
def test_reco_for_fifo(self):
# [[qty, valuation_rate, posting_date, posting_time]]
input_data = [
# [50, 1000, "2012-12-26", "12:00", 50000],
# [5, 1000, "2012-12-26", "12:00", 5000],
# [15, 1000, "2012-12-26", "12:00", 15000],
# [25, 900, "2012-12-26", "12:00", 22500],
# [20, 500, "2012-12-26", "12:00", 10000],
# [50, 1000, "2013-01-01", "12:00", 50000],
# [5, 1000, "2013-01-01", "12:00", 5000],
["", 800, "2012-12-26", "12:05", 12000],
# [20, "", "2012-12-26", "12:05", 16000]
]
for d in input_data:
self.insert_existing_sle("FIFO")
reco = self.submit_stock_reconciliation(d[0], d[1], d[2], d[3])
res = webnotes.conn.sql("""select stock_queue from `tabStock Ledger Entry`
where item_code = 'Android Jack D' and warehouse = 'Default Warehouse'
and voucher_no = 'RECO-001'""")
res = webnotes.conn.sql("""select stock_queue from `tabStock Ledger Entry`
where item_code = 'Android Jack D' and warehouse = 'Default Warehouse'
and posting_date = %s and posting_time = %s order by name desc limit 1""",
(d[2], d[3]))
stock_value = sum([v[0]*v[1] for v in json.loads(res[0][0] or "[]")])
self.assertEqual(stock_value, d[4])
self.tearDown()
self.setUp()
self.assertEqual(res[0][0], [[50, 1000]])
def test_reco_for_moving_average(self):
def atest_reco_for_moving_average(self):
webnotes.conn.set_value("Item", "Android Jack D", "valuation_method", "Moving Average")
def submit_stock_reconciliation(self, posting_date, posting_time, qty, rate):
def submit_stock_reconciliation(self, qty, rate, posting_date, posting_time):
return webnotes.model_wrapper([{
"doctype": "Stock Reconciliation",
"name": "RECO-001",
@ -61,65 +83,69 @@ class TestStockReconciliation(unittest.TestCase):
}]).submit()
def insert_test_data(self):
# create item groups and items
insert_test_data("Item Group",
sort_fn=lambda ig: (ig[0].get('parent_item_group'), ig[0].get('name')))
insert_test_data("Item")
# create default warehouse
if not webnotes.conn.exists("Warehouse", "Default Warehouse"):
webnotes.insert({"doctype": "Warehouse",
"warehouse_name": "Default Warehouse",
"warehouse_type": "Stores"})
# create UOM: Nos.
if not webnotes.conn.exists("UOM", "Nos"):
webnotes.insert({"doctype": "UOM", "uom_name": "Nos"})
# create item groups and items
insert_test_data("Item Group",
sort_fn=lambda ig: (ig[0].get('parent_item_group'), ig[0].get('name')))
insert_test_data("Item")
def insert_existing_sle(self, valuation_method):
webnotes.conn.set_value("Item", "Android Jack D", "valuation_method", valuation_method)
webnotes.conn.set_default("allow_negative_stock", 1)
existing_ledgers = [
{
"doctype": "Stock Ledger Entry", "__islocal": 1,
"voucher_type": "Stock Entry", "voucher_no": "TEST",
"item_code": "Android Jack D", "warehouse": "Default Warehouse",
"posting_date": "2012-12-12", "posting_time": "01:00:00",
"posting_date": "2012-12-12", "posting_time": "01:00",
"actual_qty": 20, "incoming_rate": 1000, "company": company
},
{
"doctype": "Stock Ledger Entry", "__islocal": 1,
"voucher_type": "Stock Entry", "voucher_no": "TEST",
"item_code": "Android Jack D", "warehouse": "Default Warehouse",
"posting_date": "2012-12-15", "posting_time": "02:00:00",
"posting_date": "2012-12-15", "posting_time": "02:00",
"actual_qty": 10, "incoming_rate": 700, "company": company
},
{
"doctype": "Stock Ledger Entry", "__islocal": 1,
"voucher_type": "Stock Entry", "voucher_no": "TEST",
"item_code": "Android Jack D", "warehouse": "Default Warehouse",
"posting_date": "2012-12-25", "posting_time": "03:00:00",
"posting_date": "2012-12-25", "posting_time": "03:00",
"actual_qty": -15, "company": company
},
{
"doctype": "Stock Ledger Entry", "__islocal": 1,
"voucher_type": "Stock Entry", "voucher_no": "TEST",
"item_code": "Android Jack D", "warehouse": "Default Warehouse",
"posting_date": "2012-12-31", "posting_time": "08:00:00",
"posting_date": "2012-12-31", "posting_time": "08:00",
"actual_qty": -20, "company": company
},
{
"doctype": "Stock Ledger Entry", "__islocal": 1,
"voucher_type": "Stock Entry", "voucher_no": "TEST",
"item_code": "Android Jack D", "warehouse": "Default Warehouse",
"posting_date": "2013-01-05", "posting_time": "07:00:00",
"posting_date": "2013-01-05", "posting_time": "07:00",
"actual_qty": 15, "incoming_rate": 1200, "company": company
},
]
pprint(webnotes.conn.sql("""select * from `tabBin` where item_code='Android Jack D'
and warehouse='Default Warehouse'""", as_dict=1))
# pprint(webnotes.conn.sql("""select * from `tabBin` where item_code='Android Jack D'
# and warehouse='Default Warehouse'""", as_dict=1))
webnotes.get_obj("Stock Ledger").update_stock(existing_ledgers)
pprint(webnotes.conn.sql("""select * from `tabBin` where item_code='Android Jack D'
and warehouse='Default Warehouse'""", as_dict=1))
# pprint(webnotes.conn.sql("""select * from `tabBin` where item_code='Android Jack D'
# and warehouse='Default Warehouse'""", as_dict=1))

29
stock/stock_ledger.py
View File

@ -40,6 +40,7 @@ def update_entries_after(args, verbose=1):
qty_after_transaction = flt(previous_sle.get("qty_after_transaction"))
valuation_rate = flt(previous_sle.get("valuation_rate"))
stock_queue = json.loads(previous_sle.get("stock_queue") or "[]")
stock_value = 0.0
entries_to_fix = get_sle_after_datetime(previous_sle or \
{"item_code": args["item_code"], "warehouse": args["warehouse"]})
@ -47,8 +48,7 @@ def update_entries_after(args, verbose=1):
valuation_method = get_valuation_method(args["item_code"])
for sle in entries_to_fix:
if sle.serial_nos or valuation_method == "FIFO" or \
not cint(webnotes.conn.get_default("allow_negative_stock")):
if sle.serial_nos or not cint(webnotes.conn.get_default("allow_negative_stock")):
# validate negative stock for serialized items, fifo valuation
# or when negative stock is not allowed for moving average
if not validate_negative_stock(qty_after_transaction, sle):
@ -75,7 +75,7 @@ def update_entries_after(args, verbose=1):
(qty_after_transaction * valuation_rate) or 0
else:
stock_value = sum((flt(batch[0]) * flt(batch[1]) for batch in stock_queue))
# update current sle
webnotes.conn.sql("""update `tabStock Ledger Entry`
set qty_after_transaction=%s, valuation_rate=%s, stock_queue=%s, stock_value=%s,
@ -83,7 +83,7 @@ def update_entries_after(args, verbose=1):
json.dumps(stock_queue), stock_value, incoming_rate, sle.name))
if _exceptions:
_raise_exceptions(args)
_raise_exceptions(args, verbose)
# update bin
webnotes.conn.sql("""update `tabBin` set valuation_rate=%s, actual_qty=%s, stock_value=%s,
@ -103,16 +103,17 @@ def get_sle_before_datetime(args):
"""
sle = get_stock_ledger_entries(args,
["timestamp(posting_date, posting_time) < timestamp(%(posting_date)s, %(posting_time)s)"],
"limit 1")
"desc", "limit 1")
return sle and sle[0] or webnotes._dict()
def get_sle_after_datetime(args):
"""get Stock Ledger Entries after a particular datetime, for reposting"""
return get_stock_ledger_entries(args,
["timestamp(posting_date, posting_time) > timestamp(%(posting_date)s, %(posting_time)s)"])
["timestamp(posting_date, posting_time) > timestamp(%(posting_date)s, %(posting_time)s)"],
"asc")
def get_stock_ledger_entries(args, conditions=None, limit=None):
def get_stock_ledger_entries(args, conditions=None, order="desc", limit=None):
"""get stock ledger entries filtered by specific posting datetime conditions"""
if not args.get("posting_date"):
args["posting_date"] = "1900-01-01"
@ -124,10 +125,11 @@ def get_stock_ledger_entries(args, conditions=None, limit=None):
and warehouse = %%(warehouse)s
and ifnull(is_cancelled, 'No') = 'No'
%(conditions)s
order by timestamp(posting_date, posting_time) desc, name desc
order by timestamp(posting_date, posting_time) %(order)s, name %(order)s
%(limit)s""" % {
"conditions": conditions and ("and " + " and ".join(conditions)) or "",
"limit": limit or ""
"limit": limit or "",
"order": order
}, args, as_dict=1)
def validate_negative_stock(qty_after_transaction, sle):
@ -202,7 +204,7 @@ def get_fifo_values(qty_after_transaction, sle, stock_queue):
if not stock_queue:
stock_queue.append([0, 0])
if actual_qty > 0:
if stock_queue[-1][0] > 0:
stock_queue.append([actual_qty, incoming_rate])
@ -213,6 +215,9 @@ def get_fifo_values(qty_after_transaction, sle, stock_queue):
incoming_cost = 0
qty_to_pop = abs(actual_qty)
while qty_to_pop:
if not stock_queue:
stock_queue.append([0, 0])
batch = stock_queue[0]
if 0 < batch[0] <= qty_to_pop:
@ -233,10 +238,10 @@ def get_fifo_values(qty_after_transaction, sle, stock_queue):
stock_qty = sum((flt(batch[0]) for batch in stock_queue))
valuation_rate = stock_qty and (stock_value / flt(stock_qty)) or 0
return valuation_rate, incoming_rate
def _raise_exceptions(args):
def _raise_exceptions(args, verbose=1):
deficiency = min(e["diff"] for e in _exceptions)
msg = """Negative stock error:
Cannot complete this transaction because stock will start