# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt

# ERPNext - web based ERP (http://erpnext.com)
# For license information, please see license.txt

import json

import frappe
from frappe.utils import add_days, cstr, flt, nowdate, nowtime

from erpnext.accounts.utils import get_stock_and_account_balance
from erpnext.stock.doctype.item.test_item import create_item
from erpnext.stock.doctype.purchase_receipt.test_purchase_receipt import make_purchase_receipt
from erpnext.stock.doctype.serial_and_batch_bundle.test_serial_and_batch_bundle import (
	get_batch_from_bundle,
	get_serial_nos_from_bundle,
	make_serial_batch_bundle,
)
from erpnext.stock.doctype.stock_reconciliation.stock_reconciliation import (
	EmptyStockReconciliationItemsError,
	get_items,
)
from erpnext.stock.doctype.warehouse.test_warehouse import create_warehouse
from erpnext.stock.stock_ledger import get_previous_sle, update_entries_after
from erpnext.stock.tests.test_utils import StockTestMixin
from erpnext.stock.utils import (
	get_combine_datetime,
	get_incoming_rate,
	get_stock_value_on,
	get_valuation_method,
)
from erpnext.tests.utils import ERPNextTestSuite


class TestStockReconciliation(ERPNextTestSuite, StockTestMixin):
	def setUp(self):
		frappe.db.set_single_value("Stock Settings", "allow_negative_stock", 1)
		frappe.local.future_sle = {}
		frappe.flags.pop("dont_execute_stock_reposts", None)

	def test_reco_for_fifo(self):
		self._test_reco_sle_gle("FIFO")

	def test_reco_for_moving_average(self):
		self._test_reco_sle_gle("Moving Average")

	@ERPNextTestSuite.change_settings("Stock Settings", {"allow_negative_stock": 1})
	def _test_reco_sle_gle(self, valuation_method):
		item_code = self.make_item(properties={"valuation_method": valuation_method}).name

		se1, se2, se3 = insert_existing_sle(warehouse="Stores - TCP1", item_code=item_code)
		company = frappe.db.get_value("Warehouse", "Stores - TCP1", "company")
		# [[qty, valuation_rate, posting_date,
		# 		posting_time, expected_stock_value, bin_qty, bin_valuation]]

		input_data = [
			[50, 1000, "2012-12-26", "12:00"],
			[25, 900, "2012-12-26", "12:00"],
			["", 1000, "2012-12-20", "12:05"],
			[20, "", "2012-12-26", "12:05"],
			[0, "", "2012-12-31", "12:10"],
		]

		for d in input_data:
			last_sle = get_previous_sle(
				{
					"item_code": item_code,
					"warehouse": "Stores - TCP1",
					"posting_date": d[2],
					"posting_time": d[3],
				}
			)

			# submit stock reconciliation
			stock_reco = create_stock_reconciliation(
				item_code=item_code,
				qty=d[0],
				rate=d[1],
				posting_date=d[2],
				posting_time=d[3],
				warehouse="Stores - TCP1",
				company=company,
				expense_account="Stock Adjustment - TCP1",
			)

			# check stock value
			sle = frappe.db.sql(
				"""select * from `tabStock Ledger Entry`
				where voucher_type='Stock Reconciliation' and voucher_no=%s""",
				stock_reco.name,
				as_dict=1,
			)

			qty_after_transaction = flt(d[0]) if d[0] != "" else flt(last_sle.get("qty_after_transaction"))

			valuation_rate = flt(d[1]) if d[1] != "" else flt(last_sle.get("valuation_rate"))

			if qty_after_transaction == last_sle.get(
				"qty_after_transaction"
			) and valuation_rate == last_sle.get("valuation_rate"):
				self.assertFalse(sle)
			else:
				self.assertEqual(flt(sle[0].qty_after_transaction, 1), flt(qty_after_transaction, 1))
				self.assertEqual(flt(sle[0].stock_value, 1), flt(qty_after_transaction * valuation_rate, 1))

				# no gl entries
				self.assertTrue(
					frappe.db.get_value(
						"Stock Ledger Entry",
						{"voucher_type": "Stock Reconciliation", "voucher_no": stock_reco.name},
					)
				)

				acc_bal, stock_bal, wh_list = get_stock_and_account_balance(
					"Stock In Hand - TCP1", stock_reco.posting_date, stock_reco.company
				)
				self.assertEqual(flt(acc_bal, 1), flt(stock_bal, 1))

				stock_reco.cancel()

		se3.cancel()
		se2.cancel()
		se1.cancel()

	def test_get_items(self):
		create_warehouse(
			"_Test Warehouse Group 1",
			{"is_group": 1, "company": "_Test Company", "parent_warehouse": "All Warehouses - _TC"},
		)
		create_warehouse(
			"_Test Warehouse Ledger 1",
			{
				"is_group": 0,
				"parent_warehouse": "_Test Warehouse Group 1 - _TC",
				"company": "_Test Company",
			},
		)

		create_item(
			"_Test Stock Reco Item",
			is_stock_item=1,
			valuation_rate=100,
			warehouse="_Test Warehouse Ledger 1 - _TC",
			opening_stock=100,
		)

		items = get_items("_Test Warehouse Group 1 - _TC", nowdate(), nowtime(), "_Test Company")

		self.assertEqual(
			["_Test Stock Reco Item", "_Test Warehouse Ledger 1 - _TC", 100],
			[items[0]["item_code"], items[0]["warehouse"], items[0]["qty"]],
		)

	def test_stock_reco_for_serialized_item(self):
		to_delete_records = []

		# Add new serial nos
		serial_item_code = "Stock-Reco-Serial-Item-1"
		serial_warehouse = "_Test Warehouse for Stock Reco1 - _TC"

		sr = create_stock_reconciliation(
			item_code=serial_item_code, warehouse=serial_warehouse, qty=5, rate=200
		)

		serial_nos = frappe.get_doc(
			"Serial and Batch Bundle", sr.items[0].serial_and_batch_bundle
		).get_serial_nos()
		self.assertEqual(len(serial_nos), 5)

		args = {
			"item_code": serial_item_code,
			"warehouse": serial_warehouse,
			"qty": -5,
			"posting_date": add_days(sr.posting_date, 1),
			"posting_time": nowtime(),
			"serial_and_batch_bundle": sr.items[0].serial_and_batch_bundle,
		}

		valuation_rate = get_incoming_rate(args)
		self.assertEqual(valuation_rate, 200)

		to_delete_records.append(sr.name)

		sr = create_stock_reconciliation(
			item_code=serial_item_code, warehouse=serial_warehouse, qty=5, rate=300, serial_no=serial_nos
		)

		sn_doc = frappe.get_doc("Serial and Batch Bundle", sr.items[0].serial_and_batch_bundle)

		self.assertEqual(len(sn_doc.get_serial_nos()), 5)

		args = {
			"item_code": serial_item_code,
			"warehouse": serial_warehouse,
			"qty": -5,
			"posting_date": add_days(sr.posting_date, 1),
			"posting_time": nowtime(),
			"serial_and_batch_bundle": sr.items[0].serial_and_batch_bundle,
		}

		valuation_rate = get_incoming_rate(args)
		self.assertEqual(valuation_rate, 300)

		to_delete_records.append(sr.name)
		to_delete_records.reverse()

		for d in to_delete_records:
			stock_doc = frappe.get_doc("Stock Reconciliation", d)
			stock_doc.cancel()

	def test_stock_reco_for_batch_item(self):
		to_delete_records = []

		# Add new serial nos
		item_code = "Stock-Reco-batch-Item-123"
		warehouse = "_Test Warehouse for Stock Reco2 - _TC"
		self.make_item(
			item_code,
			frappe._dict(
				{
					"is_stock_item": 1,
					"has_batch_no": 1,
					"create_new_batch": 1,
					"batch_number_series": "SRBI123-.#####",
				}
			),
		)

		sr = create_stock_reconciliation(
			item_code=item_code, warehouse=warehouse, qty=5, rate=200, do_not_save=1
		)
		sr.save()
		sr.submit()
		sr.load_from_db()

		batch_no = get_batch_from_bundle(sr.items[0].serial_and_batch_bundle)
		self.assertTrue(batch_no)
		to_delete_records.append(sr.name)

		sr1 = create_stock_reconciliation(
			item_code=item_code, warehouse=warehouse, qty=6, rate=300, batch_no=batch_no
		)

		args = {
			"item_code": item_code,
			"warehouse": warehouse,
			"posting_date": nowdate(),
			"posting_time": nowtime(),
			"serial_and_batch_bundle": sr1.items[0].serial_and_batch_bundle,
		}

		valuation_rate = get_incoming_rate(args)
		self.assertEqual(valuation_rate, 300)
		to_delete_records.append(sr1.name)

		sr2 = create_stock_reconciliation(
			item_code=item_code, warehouse=warehouse, qty=0, rate=0, batch_no=batch_no
		)

		stock_value = get_stock_value_on(warehouse, nowdate(), item_code)
		self.assertEqual(stock_value, 0)
		to_delete_records.append(sr2.name)

		to_delete_records.reverse()
		for d in to_delete_records:
			stock_doc = frappe.get_doc("Stock Reconciliation", d)
			stock_doc.cancel()

	def test_stock_reco_for_serial_and_batch_item(self):
		item = create_item("_TestBatchSerialItemReco")
		item.has_batch_no = 1
		item.create_new_batch = 1
		item.has_serial_no = 1
		item.batch_number_series = "TBS-BATCH-.##"
		item.serial_no_series = "TBS-.####"
		item.save()

		warehouse = "_Test Warehouse for Stock Reco2 - _TC"

		sr = create_stock_reconciliation(item_code=item.item_code, warehouse=warehouse, qty=1, rate=100)

		batch_no = get_batch_from_bundle(sr.items[0].serial_and_batch_bundle)

		serial_nos = get_serial_nos_from_bundle(sr.items[0].serial_and_batch_bundle)
		self.assertEqual(len(serial_nos), 1)
		self.assertEqual(frappe.db.get_value("Serial No", serial_nos[0], "batch_no"), batch_no)

		sr.cancel()

		self.assertEqual(frappe.db.get_value("Serial No", serial_nos[0], "warehouse"), None)

	def test_stock_reco_for_serial_and_batch_item_with_future_dependent_entry(self):
		"""
		Behaviour: 1) Create Stock Reconciliation, which will be the origin document
		of a new batch having a serial no
		2) Create a Stock Entry that adds a serial no to the same batch following this
		Stock Reconciliation
		3) Cancel Stock Entry
		Expected Result: 3) Serial No only in the Stock Entry is Inactive and Batch qty decreases
		"""
		from erpnext.stock.doctype.batch.batch import get_batch_qty
		from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry

		item = create_item("_TestBatchSerialItemDependentReco")
		item.has_batch_no = 1
		item.create_new_batch = 1
		item.has_serial_no = 1
		item.batch_number_series = "TBSD-BATCH-.##"
		item.serial_no_series = "TBSD-.####"
		item.save()

		warehouse = "_Test Warehouse for Stock Reco2 - _TC"

		stock_reco = create_stock_reconciliation(
			item_code=item.item_code, warehouse=warehouse, qty=1, rate=100
		)
		batch_no = get_batch_from_bundle(stock_reco.items[0].serial_and_batch_bundle)
		reco_serial_no = get_serial_nos_from_bundle(stock_reco.items[0].serial_and_batch_bundle)[0]

		stock_entry = make_stock_entry(
			item_code=item.item_code, target=warehouse, qty=1, basic_rate=100, batch_no=batch_no
		)
		serial_no_2 = get_serial_nos_from_bundle(stock_entry.items[0].serial_and_batch_bundle)[0]

		# Check Batch qty after 2 transactions
		batch_qty = get_batch_qty(batch_no, warehouse, item.item_code)
		self.assertEqual(batch_qty, 2)

		# Cancel latest stock document
		stock_entry.cancel()

		# Check Batch qty after cancellation
		batch_qty = get_batch_qty(batch_no, warehouse, item.item_code)
		self.assertEqual(batch_qty, 1)

		# Check if Serial No from Stock Reconcilation is intact
		self.assertEqual(frappe.db.get_value("Serial No", reco_serial_no, "batch_no"), batch_no)
		self.assertTrue(frappe.db.get_value("Serial No", reco_serial_no, "warehouse"))

		# Check if Serial No from Stock Entry is Unlinked and Inactive
		self.assertFalse(frappe.db.get_value("Serial No", serial_no_2, "warehouse"))

		stock_reco.cancel()

	def test_customer_provided_items(self):
		item_code = "Stock-Reco-customer-Item-100"
		create_item(item_code, is_customer_provided_item=1, customer="_Test Customer", is_purchase_item=0)

		sr = create_stock_reconciliation(item_code=item_code, qty=10, rate=420)

		self.assertEqual(sr.get("items")[0].allow_zero_valuation_rate, 1)
		self.assertEqual(sr.get("items")[0].valuation_rate, 0)
		self.assertEqual(sr.get("items")[0].amount, 0)

	def test_backdated_stock_reco_qty_reposting(self):
		"""
		Test if a backdated stock reco recalculates future qty until next reco.
		-------------------------------------------
		Var		| Doc	|	Qty	| Balance
		-------------------------------------------
		PR5     | PR    |   10  |  10   (posting date: today-4) [backdated]
		SR5		| Reco	|	0	|	8	(posting date: today-4) [backdated]
		PR1		| PR	|	10	|	18	(posting date: today-3)
		PR2		| PR	|	1	|	19	(posting date: today-2)
		SR4		| Reco	|	0	|	6	(posting date: today-1) [backdated]
		PR3		| PR	|	1	|	7	(posting date: today) # can't post future PR
		"""
		item_code = self.make_item().name
		warehouse = "_Test Warehouse - _TC"

		frappe.flags.dont_execute_stock_reposts = True

		def assertBalance(doc, qty_after_transaction):
			sle_balance = frappe.db.get_value(
				"Stock Ledger Entry", {"voucher_no": doc.name, "is_cancelled": 0}, "qty_after_transaction"
			)
			self.assertEqual(sle_balance, qty_after_transaction)

		pr1 = make_purchase_receipt(
			item_code=item_code, warehouse=warehouse, qty=10, rate=100, posting_date=add_days(nowdate(), -3)
		)
		pr2 = make_purchase_receipt(
			item_code=item_code, warehouse=warehouse, qty=1, rate=100, posting_date=add_days(nowdate(), -2)
		)
		pr3 = make_purchase_receipt(
			item_code=item_code, warehouse=warehouse, qty=1, rate=100, posting_date=nowdate()
		)
		assertBalance(pr1, 10)
		assertBalance(pr3, 12)

		# post backdated stock reco in between
		sr4 = create_stock_reconciliation(
			item_code=item_code, warehouse=warehouse, qty=6, rate=100, posting_date=add_days(nowdate(), -1)
		)
		assertBalance(pr3, 7)

		# post backdated stock reco at the start
		sr5 = create_stock_reconciliation(
			item_code=item_code, warehouse=warehouse, qty=8, rate=100, posting_date=add_days(nowdate(), -4)
		)
		assertBalance(pr1, 18)
		assertBalance(pr2, 19)
		assertBalance(sr4, 6)  # check if future stock reco is unaffected

		# Make a backdated receipt and check only entries till first SR are affected
		pr5 = make_purchase_receipt(
			item_code=item_code, warehouse=warehouse, qty=10, rate=100, posting_date=add_days(nowdate(), -5)
		)
		assertBalance(pr5, 10)
		# check if future stock reco is unaffected
		assertBalance(sr4, 6)
		assertBalance(sr5, 8)

		# cancel backdated stock reco and check future impact
		sr5.cancel()
		assertBalance(pr1, 10)
		assertBalance(pr2, 11)
		assertBalance(sr4, 6)  # check if future stock reco is unaffected

	@ERPNextTestSuite.change_settings("Stock Settings", {"allow_negative_stock": 0})
	def test_backdated_stock_reco_future_negative_stock(self):
		"""
		Test if a backdated stock reco causes future negative stock and is blocked.
		-------------------------------------------
		Var		| Doc	|	Qty	| Balance
		-------------------------------------------
		PR1		| PR	|	10	|	10		(posting date: today-2)
		SR3		| Reco	|	0	|	1		(posting date: today-1) [backdated & blocked]
		DN2		| DN	|	-2	|	8(-1)	(posting date: today)
		"""
		from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
		from erpnext.stock.stock_ledger import NegativeStockError

		item_code = self.make_item().name
		warehouse = "_Test Warehouse - _TC"

		pr1 = make_purchase_receipt(
			item_code=item_code, warehouse=warehouse, qty=10, rate=100, posting_date=add_days(nowdate(), -2)
		)
		dn2 = create_delivery_note(
			item_code=item_code, warehouse=warehouse, qty=2, rate=120, posting_date=nowdate()
		)

		pr1_balance = frappe.db.get_value(
			"Stock Ledger Entry", {"voucher_no": pr1.name, "is_cancelled": 0}, "qty_after_transaction"
		)
		dn2_balance = frappe.db.get_value(
			"Stock Ledger Entry", {"voucher_no": dn2.name, "is_cancelled": 0}, "qty_after_transaction"
		)
		self.assertEqual(pr1_balance, 10)
		self.assertEqual(dn2_balance, 8)

		# check if stock reco is blocked
		sr3 = create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=1,
			rate=100,
			posting_date=add_days(nowdate(), -1),
			do_not_submit=True,
		)
		self.assertRaises(NegativeStockError, sr3.submit)

		# teardown
		sr3.cancel()
		dn2.cancel()
		pr1.cancel()

	@ERPNextTestSuite.change_settings("Stock Settings", {"allow_negative_stock": 0})
	def test_backdated_stock_reco_cancellation_future_negative_stock(self):
		"""
		Test if a backdated stock reco cancellation that causes future negative stock is blocked.
		-------------------------------------------
		Var | Doc  | Qty | Balance
		-------------------------------------------
		SR  | Reco | 100 | 100     (posting date: today-1) (shouldn't be cancelled after DN)
		DN  | DN   | 100 |   0     (posting date: today)
		"""
		from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
		from erpnext.stock.stock_ledger import NegativeStockError

		item_code = self.make_item().name
		warehouse = "_Test Warehouse - _TC"

		sr = create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=100,
			rate=100,
			posting_date=add_days(nowdate(), -1),
		)

		dn = create_delivery_note(
			item_code=item_code, warehouse=warehouse, qty=100, rate=120, posting_date=nowdate()
		)

		dn_balance = frappe.db.get_value(
			"Stock Ledger Entry", {"voucher_no": dn.name, "is_cancelled": 0}, "qty_after_transaction"
		)
		self.assertEqual(dn_balance, 0)

		# check if cancellation of stock reco is blocked
		self.assertRaises(NegativeStockError, sr.cancel)

		repost_exists = bool(
			frappe.db.exists("Repost Item Valuation", {"voucher_no": sr.name, "status": "Queued"})
		)
		self.assertFalse(repost_exists, msg="Negative stock validation not working on reco cancellation")

	def test_intermediate_sr_bin_update(self):
		"""Bin should show correct qty even for backdated entries.

		-------------------------------------------
		| creation | Var | Doc  | Qty | balance qty
		-------------------------------------------
		|  1       | SR  | Reco | 10  | 10     (posting date: today+10)
		|  3       | SR2 | Reco | 11  | 11     (posting date: today+11)
		|  2       | DN  | DN   | 5   | 6 <-- assert in BIN  (posting date: today+12)
		"""
		from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note

		# repost will make this test useless, qty should update in realtime without reposts
		frappe.flags.dont_execute_stock_reposts = True

		item_code = self.make_item().name
		warehouse = "_Test Warehouse - _TC"

		create_stock_reconciliation(
			item_code=item_code, warehouse=warehouse, qty=10, rate=100, posting_date=add_days(nowdate(), 10)
		)

		create_delivery_note(
			item_code=item_code, warehouse=warehouse, qty=5, rate=120, posting_date=add_days(nowdate(), 12)
		)
		old_bin_qty = frappe.db.get_value(
			"Bin", {"item_code": item_code, "warehouse": warehouse}, "actual_qty"
		)

		create_stock_reconciliation(
			item_code=item_code, warehouse=warehouse, qty=11, rate=100, posting_date=add_days(nowdate(), 11)
		)
		new_bin_qty = frappe.db.get_value(
			"Bin", {"item_code": item_code, "warehouse": warehouse}, "actual_qty"
		)

		self.assertEqual(old_bin_qty + 1, new_bin_qty)

	def test_valid_batch(self):
		create_batch_item_with_batch("Testing Batch Item 1", "001")
		create_batch_item_with_batch("Testing Batch Item 2", "002")

		doc = frappe.get_doc(
			{
				"doctype": "Serial and Batch Bundle",
				"item_code": "Testing Batch Item 1",
				"warehouse": "_Test Warehouse - _TC",
				"voucher_type": "Stock Reconciliation",
				"entries": [
					{
						"batch_no": "002",
						"qty": 1,
						"incoming_rate": 100,
					}
				],
			}
		)

		self.assertRaises(frappe.ValidationError, doc.save)

	def test_serial_no_cancellation(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		item = create_item("Stock-Reco-Serial-Item-9", is_stock_item=1)
		if not item.has_serial_no:
			item.has_serial_no = 1
			item.serial_no_series = "PSRS9.####"
			item.save()

		item_code = item.name
		warehouse = "_Test Warehouse - _TC"

		se1 = make_stock_entry(item_code=item_code, target=warehouse, qty=10, basic_rate=700)
		serial_nos = get_serial_nos_from_bundle(se1.items[0].serial_and_batch_bundle)
		# reduce 1 item
		serial_nos.pop()
		new_serial_nos = serial_nos

		sr = create_stock_reconciliation(
			item_code=item.name, warehouse=warehouse, serial_no=new_serial_nos, qty=9
		)
		sr.cancel()

		active_sr_no = frappe.get_all(
			"Serial No", filters={"item_code": item_code, "warehouse": warehouse, "status": "Active"}
		)

		self.assertEqual(len(active_sr_no), 10)

	def test_serial_no_creation_and_inactivation(self):
		item = create_item("_TestItemCreatedWithStockReco", is_stock_item=1)
		if not item.has_serial_no:
			item.has_serial_no = 1
			item.save()

		item_code = item.name
		warehouse = "_Test Warehouse - _TC"

		if not frappe.db.exists("Serial No", "SR-CREATED-SR-NO"):
			frappe.get_doc(
				{
					"doctype": "Serial No",
					"item_code": item_code,
					"serial_no": "SR-CREATED-SR-NO",
					"company": "_Test Company",
				}
			).insert()

		sr = create_stock_reconciliation(
			item_code=item.name,
			warehouse=warehouse,
			serial_no=["SR-CREATED-SR-NO"],
			qty=1,
			do_not_submit=True,
			rate=100,
		)
		sr.save()
		self.assertEqual(cstr(sr.items[0].current_serial_no), "")
		sr.submit()

		active_sr_no = frappe.get_all(
			"Serial No", filters={"item_code": item_code, "warehouse": warehouse, "status": "Active"}
		)
		self.assertEqual(len(active_sr_no), 1)

		sr.cancel()
		active_sr_no = frappe.get_all(
			"Serial No", filters={"item_code": item_code, "warehouse": warehouse, "status": "Active"}
		)
		self.assertEqual(len(active_sr_no), 0)

	def test_serial_no_batch_no_item(self):
		item = self.make_item(
			"Test Serial No Batch No Item",
			{
				"is_stock_item": 1,
				"has_serial_no": 1,
				"has_batch_no": 1,
				"serial_no_series": "SRS9.####",
				"batch_number_series": "BNS90.####",
				"create_new_batch": 1,
			},
		)

		warehouse = "_Test Warehouse - _TC"

		sr = create_stock_reconciliation(
			item_code=item.name,
			warehouse=warehouse,
			qty=1,
			rate=100,
		)

		sl_entry = frappe.db.get_value(
			"Stock Ledger Entry",
			{"voucher_type": "Stock Reconciliation", "voucher_no": sr.name},
			["actual_qty", "qty_after_transaction"],
			as_dict=1,
		)

		self.assertEqual(flt(sl_entry.actual_qty), 1.0)
		self.assertEqual(flt(sl_entry.qty_after_transaction), 1.0)

	@ERPNextTestSuite.change_settings("Stock Reposting Settings", {"item_based_reposting": 0})
	def test_backdated_stock_reco_entry(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		item_code = self.make_item(
			"Test New Batch Item ABCV",
			{
				"is_stock_item": 1,
				"has_batch_no": 1,
				"batch_number_series": "BNS91.####",
				"create_new_batch": 1,
			},
		).name

		warehouse = "_Test Warehouse - _TC"

		# Added 100 Qty, Balace Qty 100
		se1 = make_stock_entry(
			item_code=item_code, posting_time="09:00:00", target=warehouse, qty=100, basic_rate=700
		)

		batch_no = get_batch_from_bundle(se1.items[0].serial_and_batch_bundle)

		# Removed 50 Qty, Balace Qty 50
		make_stock_entry(
			item_code=item_code,
			batch_no=batch_no,
			posting_time="10:00:00",
			source=warehouse,
			qty=50,
			basic_rate=700,
		)

		# Stock Reco for 100, Balace Qty 100
		stock_reco = create_stock_reconciliation(
			item_code=item_code,
			posting_time="11:00:00",
			warehouse=warehouse,
			batch_no=batch_no,
			qty=100,
			rate=100,
		)

		stock_reco.reload()
		stock_reco_sabb = stock_reco.items[0].serial_and_batch_bundle
		posting_datetime = frappe.db.get_value("Serial and Batch Bundle", stock_reco_sabb, "posting_datetime")
		self.assertEqual(
			posting_datetime, get_combine_datetime(stock_reco.posting_date, stock_reco.posting_time)
		)

		sle = frappe.get_all(
			"Stock Ledger Entry",
			filters={"is_cancelled": 0, "voucher_no": stock_reco.name, "actual_qty": ("<", 0)},
			fields=["actual_qty"],
		)

		self.assertEqual(flt(sle[0].actual_qty), flt(-50.0))

		# Removed 50 Qty, Balace Qty 50
		make_stock_entry(
			item_code=item_code,
			batch_no=batch_no,
			posting_time="12:00:00",
			source=warehouse,
			qty=52,
			basic_rate=700,
		)

		self.assertFalse(frappe.db.exists("Repost Item Valuation", {"voucher_no": stock_reco.name}))

		self.assertRaises(frappe.ValidationError, stock_reco.cancel)

	def test_update_stock_reconciliation_while_reposting(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		item_code = self.make_item().name
		warehouse = "_Test Warehouse - _TC"

		# Stock Value => 100 * 100 = 10000
		make_stock_entry(
			item_code=item_code,
			target=warehouse,
			qty=100,
			basic_rate=100,
			posting_time="10:00:00",
		)

		# Stock Value => 100 * 200 = 20000
		# Value Change => 20000 - 10000 = 10000
		sr1 = create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=100,
			rate=200,
			posting_time="12:00:00",
		)
		self.assertEqual(sr1.difference_amount, 10000)

		# Stock Value => 50 * 50 = 2500
		# Value Change => 2500 - 10000 = -7500
		sr2 = create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=50,
			rate=50,
			posting_time="11:00:00",
		)
		self.assertEqual(sr2.difference_amount, -7500)

		sr1.load_from_db()
		self.assertEqual(sr1.difference_amount, 17500)

		sr2.cancel()
		sr1.load_from_db()
		self.assertEqual(sr1.difference_amount, 10000)

	def assert_reco_difference_matches_gl(self, reco_name):
		"""The displayed Difference Amount (doc and per-row) must equal the reposted GL impact,
		i.e. the sum of the reconciliation's Stock Ledger Entry ``stock_value_difference``."""
		from erpnext.stock.doctype.stock_reconciliation.stock_reconciliation import (
			get_row_stock_value_difference,
		)

		reco = frappe.get_doc("Stock Reconciliation", reco_name)
		total_difference = 0.0

		for row in reco.items:
			row_difference = flt(
				get_row_stock_value_difference("Stock Reconciliation", reco_name, row.name),
				row.precision("amount_difference"),
			)

			self.assertEqual(flt(row.amount_difference), row_difference)
			total_difference += row_difference

		self.assertEqual(
			flt(reco.difference_amount, reco.precision("difference_amount")),
			flt(total_difference, reco.precision("difference_amount")),
		)

	def test_difference_amount_synced_with_gl_after_repost_non_serialized(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		item_code = self.make_item().name
		warehouse = "_Test Warehouse - _TC"

		# Opening stock => 100 * 100 = 10000
		make_stock_entry(
			item_code=item_code,
			target=warehouse,
			qty=100,
			basic_rate=100,
			posting_date=add_days(nowdate(), -5),
			posting_time="10:00:00",
		)

		# Reconcile to 100 @ 200 => difference 20000 - 10000 = 10000
		reco = create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=100,
			rate=200,
			posting_date=add_days(nowdate(), -2),
		)
		self.assertEqual(reco.difference_amount, 10000)
		self.assert_reco_difference_matches_gl(reco.name)

		# Backdated reconciliation lowers the pre-reco stock value to 50 * 50 = 2500
		create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=50,
			rate=50,
			posting_date=add_days(nowdate(), -3),
		)

		reco.load_from_db()
		# Current is now 2500 => difference 20000 - 2500 = 17500
		self.assertEqual(reco.difference_amount, 17500)
		self.assert_reco_difference_matches_gl(reco.name)

	def test_difference_amount_synced_with_gl_after_repost_batched(self):
		from erpnext.stock.doctype.landed_cost_voucher.test_landed_cost_voucher import (
			make_landed_cost_voucher,
		)

		item_code = self.make_item(
			"Test Batch Item Reco Difference Sync",
			{
				"is_stock_item": 1,
				"has_batch_no": 1,
				"create_new_batch": 1,
				"batch_number_series": "TEST-BATCH-DIFFSYNC-.###",
			},
		).name
		warehouse = "_Test Warehouse - _TC"

		# Receive 10 @ 100 (batch value 1000)
		pr = make_purchase_receipt(
			item_code=item_code,
			warehouse=warehouse,
			qty=10,
			rate=100,
			posting_date=add_days(nowdate(), -5),
		)
		batch_no = get_batch_from_bundle(pr.items[0].serial_and_batch_bundle)

		# Reconcile the batch to 10 @ 500 => difference 5000 - 1000 = 4000
		reco = create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=10,
			rate=500,
			batch_no=batch_no,
			use_serial_batch_fields=1,
			posting_date=add_days(nowdate(), -2),
		)
		difference_on_submit = reco.difference_amount
		self.assert_reco_difference_matches_gl(reco.name)

		# Landed cost retroactively raises the receipt (and batch) valuation, reposting the reco
		make_landed_cost_voucher(
			receipt_document_type="Purchase Receipt",
			receipt_document=pr.name,
			charges=1000,
			company="_Test Company",
		)

		reco.load_from_db()
		self.assertNotEqual(reco.difference_amount, difference_on_submit)
		self.assert_reco_difference_matches_gl(reco.name)

	def test_difference_amount_synced_with_gl_after_repost_serialized(self):
		from erpnext.stock.doctype.landed_cost_voucher.test_landed_cost_voucher import (
			make_landed_cost_voucher,
		)

		item_code = self.make_item(
			"Test Serial Item Reco Difference Sync",
			{
				"is_stock_item": 1,
				"has_serial_no": 1,
				"serial_no_series": "TSIRDS.####",
			},
		).name
		warehouse = "_Test Warehouse - _TC"

		# Receive 5 serial nos @ 100 (value 500)
		pr = make_purchase_receipt(
			item_code=item_code,
			warehouse=warehouse,
			qty=5,
			rate=100,
			posting_date=add_days(nowdate(), -5),
		)
		serial_nos = get_serial_nos_from_bundle(pr.items[0].serial_and_batch_bundle)

		# Reconcile the serial nos to 5 @ 500 => difference 2500 - 500 = 2000
		reco = create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=5,
			rate=500,
			serial_no="\n".join(serial_nos),
			use_serial_batch_fields=1,
			posting_date=add_days(nowdate(), -2),
		)
		difference_on_submit = reco.difference_amount
		self.assert_reco_difference_matches_gl(reco.name)

		# Landed cost retroactively raises the receipt (and serial) valuation, reposting the reco
		make_landed_cost_voucher(
			receipt_document_type="Purchase Receipt",
			receipt_document=pr.name,
			charges=1000,
			company="_Test Company",
		)

		reco.load_from_db()
		self.assertNotEqual(reco.difference_amount, difference_on_submit)
		self.assert_reco_difference_matches_gl(reco.name)

	def test_make_stock_zero_for_serial_batch_item(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		serial_item = self.make_item(
			properties={"is_stock_item": 1, "has_serial_no": 1, "serial_no_series": "DJJ.####"}
		).name
		batch_item = self.make_item(
			properties={
				"is_stock_item": 1,
				"has_batch_no": 1,
				"batch_number_series": "BDJJ.####",
				"create_new_batch": 1,
			}
		).name

		serial_batch_item = self.make_item(
			properties={
				"is_stock_item": 1,
				"has_batch_no": 1,
				"batch_number_series": "ADJJ.####",
				"create_new_batch": 1,
				"has_serial_no": 1,
				"serial_no_series": "SN-ADJJ.####",
			}
		).name

		warehouse = "_Test Warehouse - _TC"

		for item_code in [serial_item, batch_item, serial_batch_item]:
			make_stock_entry(
				item_code=item_code,
				target=warehouse,
				qty=10,
				basic_rate=100,
			)

			_reco = create_stock_reconciliation(
				item_code=item_code,
				warehouse=warehouse,
				qty=0.0,
			)

			serial_batch_bundle = frappe.get_all(
				"Stock Ledger Entry",
				{"item_code": item_code, "warehouse": warehouse, "is_cancelled": 0, "voucher_no": _reco.name},
				"serial_and_batch_bundle",
			)

			self.assertEqual(len(serial_batch_bundle), 1)

			_reco.cancel()

			serial_batch_bundle = frappe.get_all(
				"Stock Ledger Entry",
				{"item_code": item_code, "warehouse": warehouse, "is_cancelled": 0, "voucher_no": _reco.name},
				"serial_and_batch_bundle",
			)

			self.assertEqual(len(serial_batch_bundle), 0)

	def test_backdated_purchase_receipt_with_stock_reco(self):
		item_code = self.make_item(
			properties={
				"is_stock_item": 1,
				"has_serial_no": 1,
				"serial_no_series": "TEST-SERIAL-.###",
			}
		).name

		warehouse = "_Test Warehouse - _TC"

		# Step - 1: Create a Backdated Purchase Receipt

		pr1 = make_purchase_receipt(
			item_code=item_code, warehouse=warehouse, qty=10, rate=100, posting_date=add_days(nowdate(), -3)
		)
		pr1.reload()

		serial_nos = sorted(get_serial_nos_from_bundle(pr1.items[0].serial_and_batch_bundle))[:5]

		# Step - 2: Create a Stock Reconciliation
		sr1 = create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=5,
			serial_no=serial_nos,
		)

		data = frappe.get_all(
			"Stock Ledger Entry",
			fields=["serial_no", "actual_qty", "stock_value_difference"],
			filters={"voucher_no": sr1.name, "is_cancelled": 0},
			order_by="creation",
		)

		for d in data:
			if d.actual_qty < 0:
				self.assertEqual(d.actual_qty, -10.0)
				self.assertAlmostEqual(d.stock_value_difference, -1000.0)
			else:
				self.assertEqual(d.actual_qty, 5.0)
				self.assertAlmostEqual(d.stock_value_difference, 500.0)

		# Step - 3: Create a Purchase Receipt before the first Purchase Receipt
		pr = make_purchase_receipt(
			item_code=item_code,
			warehouse=warehouse,
			qty=10,
			rate=200,
			posting_date=add_days(nowdate(), -5),
			do_not_submit=True,
		)

		self.assertRaises(frappe.ValidationError, pr.submit)

	def test_balance_qty_for_batch_with_backdated_stock_reco_and_future_entries(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		item = self.make_item(
			"Test Batch Item Original Test",
			{
				"is_stock_item": 1,
				"has_batch_no": 1,
				"create_new_batch": 1,
				"batch_number_series": "TEST-BATCH-SRWFEE-.###",
			},
		)

		warehouse = "_Test Warehouse - _TC"
		se1 = make_stock_entry(
			item_code=item.name,
			target=warehouse,
			qty=50,
			basic_rate=100,
			posting_date=add_days(nowdate(), -2),
		)
		batch1 = get_batch_from_bundle(se1.items[0].serial_and_batch_bundle)

		se2 = make_stock_entry(
			item_code=item.name,
			target=warehouse,
			qty=50,
			basic_rate=100,
			posting_date=add_days(nowdate(), -2),
		)
		batch2 = get_batch_from_bundle(se2.items[0].serial_and_batch_bundle)

		se3 = make_stock_entry(
			item_code=item.name,
			target=warehouse,
			qty=100,
			basic_rate=100,
			posting_date=add_days(nowdate(), -2),
		)
		batch3 = get_batch_from_bundle(se3.items[0].serial_and_batch_bundle)

		se3 = make_stock_entry(
			item_code=item.name,
			target=warehouse,
			qty=100,
			basic_rate=100,
			posting_date=nowdate(),
		)

		sle = frappe.get_all(
			"Stock Ledger Entry",
			filters={
				"item_code": item.name,
				"warehouse": warehouse,
				"is_cancelled": 0,
				"voucher_no": se3.name,
			},
			fields=["qty_after_transaction"],
			order_by="posting_time desc, creation desc",
		)

		self.assertEqual(flt(sle[0].qty_after_transaction), flt(300.0))

		sr = create_stock_reconciliation(
			item_code=item.name,
			warehouse=warehouse,
			qty=0,
			batch_no=batch1,
			posting_date=add_days(nowdate(), -1),
			use_serial_batch_fields=1,
			do_not_save=1,
		)

		for batch in [batch2, batch3]:
			sr.append(
				"items",
				{
					"item_code": item.name,
					"warehouse": warehouse,
					"qty": 0,
					"batch_no": batch,
					"use_serial_batch_fields": 1,
				},
			)

		sr.save()
		sr.submit()

		sle = frappe.get_all(
			"Stock Ledger Entry",
			filters={
				"item_code": item.name,
				"warehouse": warehouse,
				"is_cancelled": 0,
				"voucher_no": se3.name,
			},
			fields=["qty_after_transaction"],
			order_by="posting_time desc, creation desc",
		)

		self.assertEqual(flt(sle[0].qty_after_transaction), flt(100.0))

	def test_stock_reco_and_backdated_purchase_receipt(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		item = self.make_item(
			"Test Batch Item Original STOCK RECO Test",
			{
				"is_stock_item": 1,
				"has_batch_no": 1,
				"create_new_batch": 1,
				"batch_number_series": "TEST-BATCH-SRCOSRWFEE-.###",
			},
		)

		warehouse = "_Test Warehouse - _TC"

		sr = create_stock_reconciliation(
			item_code=item.name,
			warehouse=warehouse,
			qty=100,
			rate=100,
		)

		sr.reload()
		self.assertTrue(sr.items[0].serial_and_batch_bundle)
		self.assertFalse(sr.items[0].current_serial_and_batch_bundle)
		batch = get_batch_from_bundle(sr.items[0].serial_and_batch_bundle)

		se1 = make_stock_entry(
			item_code=item.name,
			target=warehouse,
			qty=50,
			basic_rate=100,
			posting_date=add_days(nowdate(), -2),
		)

		batch1 = get_batch_from_bundle(se1.items[0].serial_and_batch_bundle)
		self.assertFalse(batch1 == batch)

		sr.reload()
		self.assertTrue(sr.items[0].serial_and_batch_bundle)
		self.assertFalse(sr.items[0].current_serial_and_batch_bundle)

	def test_not_reconcile_all_batch(self):
		from erpnext.stock.doctype.batch.batch import get_batch_qty
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		item = self.make_item(
			"Test Batch Item Not Reconcile All Serial Batch",
			{
				"is_stock_item": 1,
				"has_batch_no": 1,
				"create_new_batch": 1,
				"batch_number_series": "TEST-BATCH-NRALL-SRCOSRWFEE-.###",
			},
		)

		warehouse = "_Test Warehouse - _TC"

		batches = []
		for qty in [10, 20, 30]:
			se = make_stock_entry(
				item_code=item.name,
				target=warehouse,
				qty=qty,
				basic_rate=100 + qty,
				posting_date=nowdate(),
			)

			batch_no = get_batch_from_bundle(se.items[0].serial_and_batch_bundle)
			batches.append(frappe._dict({"batch_no": batch_no, "qty": qty}))

		sr = create_stock_reconciliation(
			item_code=item.name,
			warehouse=warehouse,
			qty=100,
			rate=1000,
			reconcile_all_serial_batch=0,
			batch_no=batches[0].batch_no,
		)

		sr.reload()
		self.assertEqual(sr.difference_amount, 98900.0)

		self.assertTrue(sr.items[0].current_valuation_rate)
		current_sabb = sr.items[0].current_serial_and_batch_bundle
		doc = frappe.get_doc("Serial and Batch Bundle", current_sabb)
		for row in doc.entries:
			self.assertEqual(row.batch_no, batches[0].batch_no)
			self.assertEqual(row.qty, batches[0].qty * -1)

		batch_qty = get_batch_qty(batches[0].batch_no, warehouse, item.name)
		self.assertEqual(batch_qty, 100)

		for row in frappe.get_all("Repost Item Valuation", filters={"voucher_no": sr.name}):
			rdoc = frappe.get_doc("Repost Item Valuation", row.name)
			rdoc.cancel()
			rdoc.delete()

		sr.cancel()

		for row in frappe.get_all(
			"Serial and Batch Bundle", fields=["docstatus"], filters={"voucher_no": sr.name}
		):
			self.assertEqual(row.docstatus, 2)

	def test_not_reconcile_all_serial_nos(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry
		from erpnext.stock.utils import get_incoming_rate

		item = self.make_item(
			"Test Serial NO Item Not Reconcile All Serial Batch",
			{
				"is_stock_item": 1,
				"has_serial_no": 1,
				"serial_no_series": "SNN-TEST-BATCH-NRALL-S-.###",
			},
		)

		warehouse = "_Test Warehouse - _TC"

		serial_nos = []
		for qty in [5, 5, 5]:
			se = make_stock_entry(
				item_code=item.name,
				target=warehouse,
				qty=qty,
				basic_rate=100 + qty,
				posting_date=nowdate(),
			)

			serial_nos.extend(get_serial_nos_from_bundle(se.items[0].serial_and_batch_bundle))

		sr = create_stock_reconciliation(
			item_code=item.name,
			warehouse=warehouse,
			qty=5,
			rate=1000,
			reconcile_all_serial_batch=0,
			serial_no=serial_nos[0:5],
		)

		sr.reload()
		current_sabb = sr.items[0].current_serial_and_batch_bundle
		doc = frappe.get_doc("Serial and Batch Bundle", current_sabb)
		for row in doc.entries:
			self.assertEqual(row.serial_no, serial_nos[row.idx - 1])

		sabb = sr.items[0].serial_and_batch_bundle
		doc = frappe.get_doc("Serial and Batch Bundle", sabb)
		for row in doc.entries:
			self.assertEqual(row.qty, 1)
			self.assertAlmostEqual(row.incoming_rate, 1000.00)
			self.assertEqual(row.serial_no, serial_nos[row.idx - 1])

	def test_stock_reco_with_legacy_batch(self):
		from erpnext.stock.doctype.batch.batch import get_batch_qty

		batch_item_code = self.make_item(
			"Test Batch Item Legacy Batch 1",
			{
				"is_stock_item": 1,
				"has_batch_no": 1,
				"create_new_batch": 1,
				"batch_number_series": "BH1-NRALL-S-.###",
			},
		).name

		warehouse = "_Test Warehouse - _TC"

		frappe.flags.ignore_serial_batch_bundle_validation = True
		frappe.flags.use_serial_and_batch_fields = True

		batch_id = "BH1-NRALL-S-0001"
		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)

		stock_queue = []
		qty_after_transaction = 0
		balance_value = 0
		i = 0
		for qty, valuation in {10: 100, 20: 200}.items():
			i += 1
			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": add_days(nowdate(), -2 * i),
					"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.set_posting_datetime()
			doc.flags.ignore_permissions = True
			doc.flags.ignore_mandatory = True
			doc.flags.ignore_links = True
			doc.flags.ignore_validate = True
			doc.submit()
			doc.reload()

		frappe.flags.ignore_serial_batch_bundle_validation = False
		frappe.flags.use_serial_and_batch_fields = False

		batch_doc = frappe.get_doc("Batch", batch_id)

		qty = get_batch_qty(batch_id, warehouse, batch_item_code)
		self.assertEqual(qty, 30)

		sr = create_stock_reconciliation(
			item_code=batch_item_code,
			posting_date=add_days(nowdate(), -3),
			posting_time=nowtime(),
			warehouse=warehouse,
			qty=100,
			rate=1000,
			reconcile_all_serial_batch=0,
			batch_no=batch_id,
			use_serial_batch_fields=1,
		)

		self.assertEqual(sr.items[0].current_qty, 20)
		self.assertEqual(sr.items[0].qty, 100)

		qty = get_batch_qty(batch_id, warehouse, batch_item_code)
		self.assertEqual(qty, 110)

	def test_skip_reposting_for_entries_after_stock_reco(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		item_code = create_item("Test Item For Skip Reposting After Stock Reco", is_stock_item=1).name

		warehouse = "_Test Warehouse - _TC"

		make_stock_entry(
			posting_date="2024-11-01",
			posting_time="11:00",
			item_code=item_code,
			target=warehouse,
			qty=10,
			basic_rate=100,
		)

		create_stock_reconciliation(
			posting_date="2024-11-02",
			posting_time="11:00",
			item_code=item_code,
			warehouse=warehouse,
			qty=20,
			rate=100,
		)

		se = make_stock_entry(
			posting_date="2024-11-03",
			posting_time="11:00",
			item_code=item_code,
			source=warehouse,
			qty=15,
		)

		stock_value_difference = frappe.db.get_value(
			"Stock Ledger Entry", {"voucher_no": se.name, "is_cancelled": 0}, "stock_value_difference"
		)

		self.assertEqual(stock_value_difference, 1500.00 * -1)

		make_stock_entry(
			posting_date="2024-10-29",
			posting_time="11:00",
			item_code=item_code,
			target=warehouse,
			qty=10,
			basic_rate=100,
		)

		stock_value_difference = frappe.db.get_value(
			"Stock Ledger Entry", {"voucher_no": se.name, "is_cancelled": 0}, "stock_value_difference"
		)

		self.assertEqual(stock_value_difference, 1500.00 * -1)

	def test_stock_reco_for_negative_batch(self):
		from erpnext.stock.doctype.batch.batch import get_batch_qty
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		item_code = self.make_item(
			"Test Item For Negative Batch",
			{
				"is_stock_item": 1,
				"has_batch_no": 1,
				"create_new_batch": 1,
				"batch_number_series": "TEST-BATCH-NB-.###",
			},
		).name

		warehouse = "_Test Warehouse - _TC"

		se = make_stock_entry(
			posting_date="2024-11-01",
			posting_time="11:00",
			item_code=item_code,
			target=warehouse,
			qty=10,
			basic_rate=100,
		)

		batch_no = get_batch_from_bundle(se.items[0].serial_and_batch_bundle)

		se = make_stock_entry(
			posting_date="2024-11-01",
			posting_time="11:00",
			item_code=item_code,
			source=warehouse,
			qty=10,
			basic_rate=100,
			use_serial_batch_fields=1,
			batch_no=batch_no,
		)

		sles = frappe.get_all(
			"Stock Ledger Entry",
			filters={"voucher_no": se.name, "is_cancelled": 0},
		)

		# intentionally setting negative qty
		doc = frappe.get_doc("Stock Ledger Entry", sles[0].name)
		doc.db_set(
			{
				"actual_qty": -20,
				"qty_after_transaction": -10,
			}
		)

		sabb_doc = frappe.get_doc("Serial and Batch Bundle", doc.serial_and_batch_bundle)
		for row in sabb_doc.entries:
			row.db_set("qty", -20)

		batch_qty = get_batch_qty(batch_no, warehouse, item_code, consider_negative_batches=True)
		self.assertEqual(batch_qty, -10)

		sr = create_stock_reconciliation(
			posting_date="2024-11-02",
			posting_time="11:00",
			item_code=item_code,
			warehouse=warehouse,
			use_serial_batch_fields=1,
			batch_no=batch_no,
			qty=0,
			rate=100,
			do_not_submit=True,
		)

		self.assertEqual(sr.items[0].current_qty, -10)
		sr.submit()
		sr.reload()

		self.assertTrue(sr.items[0].current_serial_and_batch_bundle)
		self.assertFalse(sr.items[0].serial_and_batch_bundle)

	def test_stock_reco_batch_item_current_valuation(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		# Add new serial nos
		item_code = "Stock-Reco-batch-Item-1234"
		warehouse = "_Test Warehouse - _TC"
		self.make_item(
			item_code,
			frappe._dict(
				{
					"is_stock_item": 1,
					"has_batch_no": 1,
					"create_new_batch": 1,
					"batch_number_series": "JJ-SRI1234-.#####",
				}
			),
		)

		se = make_stock_entry(
			item_code=item_code,
			target=warehouse,
			qty=1,
			basic_rate=100,
		)

		batch_no = get_batch_from_bundle(se.items[0].serial_and_batch_bundle)

		sr = create_stock_reconciliation(
			item_code=item_code, warehouse=warehouse, qty=0, rate=100, do_not_save=1
		)

		sr.items[0].batch_no = batch_no
		sr.items[0].use_serial_batch_fields = 1
		sr.save()
		self.assertEqual(sr.items[0].current_valuation_rate, 100)
		self.assertEqual(sr.difference_amount, 100 * -1)
		self.assertTrue(sr.items[0].qty == 0)

	def test_stock_reco_recalculate_qty_for_backdated_entry(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		item_code = self.make_item(
			"Test Batch Item Stock Reco Recalculate Qty",
			{
				"is_stock_item": 1,
				"has_batch_no": 1,
				"create_new_batch": 1,
				"batch_number_series": "TEST-BATCH-RRQ-.###",
			},
		).name

		warehouse = "_Test Warehouse - _TC"

		sr = create_stock_reconciliation(
			item_code=item_code,
			posting_date=add_days(nowdate(), -2),
			warehouse=warehouse,
			qty=10,
			rate=100,
			use_serial_batch_fields=1,
			purpose="Opening Stock",
		)

		sr.reload()
		self.assertEqual(sr.items[0].current_qty, 0)
		self.assertEqual(sr.items[0].current_valuation_rate, 0)

		batch_no = get_batch_from_bundle(sr.items[0].serial_and_batch_bundle)
		stock_ledgers = frappe.get_all(
			"Stock Ledger Entry",
			filters={"voucher_no": sr.name, "is_cancelled": 0},
			pluck="name",
		)

		self.assertTrue(len(stock_ledgers) == 1)

		se = make_stock_entry(
			item_code=item_code,
			source=warehouse,
			qty=10,
			basic_rate=100,
			use_serial_batch_fields=1,
			batch_no=batch_no,
		)

		# Make backdated stock reconciliation entry
		create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=10,
			rate=200,
			use_serial_batch_fields=1,
			batch_no=batch_no,
			posting_date=add_days(nowdate(), -1),
		)

		stock_ledger = frappe.get_all(
			"Stock Ledger Entry",
			filters={"voucher_no": se.name, "is_cancelled": 0},
			fields=["stock_value_difference"],
		)

		self.assertEqual(stock_ledger[0].stock_value_difference, 2000.0 * -1)

	def test_serial_no_backdated_stock_reco(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		serial_item = self.make_item(
			"Test Serial Item Stock Reco Backdated",
			{
				"is_stock_item": 1,
				"has_serial_no": 1,
				"serial_no_series": "TSISRB.####",
			},
		).name

		warehouse = "_Test Warehouse - _TC"

		se = make_stock_entry(
			item_code=serial_item,
			target=warehouse,
			qty=1,
			basic_rate=100,
			use_serial_batch_fields=1,
		)

		serial_no = get_serial_nos_from_bundle(se.items[0].serial_and_batch_bundle)[0]
		status = frappe.get_value(
			"Serial No",
			serial_no,
			"status",
		)

		self.assertTrue(status == "Active")

		sr = create_stock_reconciliation(
			item_code=serial_item,
			warehouse=warehouse,
			qty=1,
			rate=200,
			use_serial_batch_fields=1,
			serial_no=serial_no,
		)

		sr.reload()

		status = frappe.get_value(
			"Serial No",
			serial_no,
			"status",
		)

		self.assertTrue(status == "Active")

		se = make_stock_entry(
			item_code=serial_item,
			source=warehouse,
			qty=1,
			basic_rate=100,
			use_serial_batch_fields=1,
		)

		status = frappe.get_value(
			"Serial No",
			serial_no,
			"status",
		)

		self.assertFalse(status == "Active")

		sr.cancel()

		status = frappe.get_value(
			"Serial No",
			serial_no,
			"status",
		)

		self.assertFalse(status == "Active")

	def test_change_valuation_of_batch_using_backdated_stock_reco(self):
		from erpnext.stock.doctype.batch.batch import get_batch_qty
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

		item_code = self.make_item(
			"Test Item Change Valuation of Batch",
			{
				"is_stock_item": 1,
				"has_batch_no": 1,
				"create_new_batch": 1,
				"batch_number_series": "TEST-BATCH-CVB-.###",
			},
		).name

		warehouse = "_Test Warehouse - _TC"

		reco = create_stock_reconciliation(
			item_code=item_code,
			posting_date=add_days(nowdate(), -6),
			warehouse=warehouse,
			qty=10,
			rate=80,
			use_serial_batch_fields=1,
			purpose="Opening Stock",
		)

		batch_no = get_batch_from_bundle(reco.items[0].serial_and_batch_bundle)

		make_stock_entry(
			item_code=item_code,
			source=warehouse,
			qty=2,
			posting_date=add_days(nowdate(), -4),
			use_serial_batch_fields=1,
			batch_no=batch_no,
		)

		make_stock_entry(
			item_code=item_code,
			source=warehouse,
			qty=2,
			posting_date=add_days(nowdate(), -3),
			use_serial_batch_fields=1,
			batch_no=batch_no,
		)

		se4 = make_stock_entry(
			item_code=item_code,
			source=warehouse,
			qty=2,
			posting_date=add_days(nowdate(), -2),
			use_serial_batch_fields=1,
			batch_no=batch_no,
		)

		sle = frappe.db.get_value(
			"Stock Ledger Entry",
			{"voucher_no": se4.name, "is_cancelled": 0},
			["actual_qty", "stock_value_difference"],
			as_dict=1,
		)

		valuation_rate = sle.stock_value_difference / sle.actual_qty
		self.assertEqual(valuation_rate, 80)

		create_stock_reconciliation(
			item_code=item_code,
			posting_date=add_days(nowdate(), -5),
			warehouse=warehouse,
			qty=10,
			rate=100,
			use_serial_batch_fields=1,
			batch_no=batch_no,
		)

		sle = frappe.db.get_value(
			"Stock Ledger Entry",
			{"voucher_no": se4.name, "is_cancelled": 0},
			["actual_qty", "stock_value_difference"],
			as_dict=1,
		)

		valuation_rate = sle.stock_value_difference / sle.actual_qty

		self.assertEqual(valuation_rate, 100)

		batch_qty = get_batch_qty(batch_no, warehouse, item_code)
		self.assertEqual(batch_qty, 4)

	def test_sabb_cancel_on_stock_reco_cancellation(self):
		item_code = self.make_item(
			"Test Item for SABB Cancel on Stock Reco Cancellation",
			{
				"is_stock_item": 1,
				"has_batch_no": 1,
				"create_new_batch": 1,
				"batch_number_series": "TEST-BATCH-SABBCANC-.###",
			},
		).name

		warehouse = "_Test Warehouse - _TC"

		sr = create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=10,
			rate=100,
			use_serial_batch_fields=1,
			purpose="Opening Stock",
		)

		sr.reload()

		batch_no = get_batch_from_bundle(sr.items[0].serial_and_batch_bundle)

		sr1 = create_stock_reconciliation(
			item_code=item_code,
			warehouse=warehouse,
			qty=20,
			rate=100,
			use_serial_batch_fields=1,
			batch_no=batch_no,
		)

		sr1.reload()

		current_serial_and_batch_bundle = sr1.items[0].current_serial_and_batch_bundle
		serial_and_batch_bundle = sr1.items[0].serial_and_batch_bundle

		self.assertTrue(current_serial_and_batch_bundle)
		self.assertTrue(serial_and_batch_bundle)

		sr1.cancel()

		for sabb in [serial_and_batch_bundle, current_serial_and_batch_bundle]:
			docstatus = frappe.db.get_value(
				"Serial and Batch Bundle",
				sabb,
				"docstatus",
			)

			self.assertEqual(docstatus, 2)

	def test_stock_reco_with_opening_stock_with_diff_inventory(self):
		from erpnext.stock.doctype.inventory_dimension.test_inventory_dimension import (
			create_inventory_dimension,
		)

		if frappe.db.exists("DocType", "Plant"):
			return

		doctype = frappe.get_doc(
			{
				"doctype": "DocType",
				"name": "Plant",
				"module": "Stock",
				"custom": 1,
				"fields": [
					{
						"fieldname": "plant_name",
						"fieldtype": "Data",
						"label": "Plant Name",
						"reqd": 1,
					}
				],
				"autoname": "field:plant_name",
			}
		)
		doctype.insert(ignore_permissions=True)
		create_inventory_dimension(dimension_name="ID-Plant", reference_document="Plant")

		plant_a = frappe.get_doc(
			{
				"doctype": "Plant",
				"plant_name": "Plant A",
			}
		).insert(ignore_permissions=True)

		plant_b = frappe.get_doc(
			{
				"doctype": "Plant",
				"plant_name": "Plant B",
			}
		).insert(ignore_permissions=True)

		warehouse = "_Test Warehouse - _TC"

		item_code = "Item-Test"
		item = self.make_item(item_code, {"is_stock_item": 1})

		sr = frappe.new_doc("Stock Reconciliation")
		sr.purpose = "Opening Stock"
		sr.posting_date = nowdate()
		sr.posting_time = nowtime()
		sr.company = "_Test Company"

		sr.append(
			"items",
			{
				"item_code": item.name,
				"warehouse": warehouse,
				"qty": 5,
				"valuation_rate": 100,
				"id_plant": plant_a.name,
			},
		)

		sr.append(
			"items",
			{
				"item_code": item.name,
				"warehouse": warehouse,
				"qty": 3,
				"valuation_rate": 110,
				"id_plant": plant_b.name,
			},
		)

		sr.insert()
		sr.submit()

		self.assertEqual(len(sr.items), 2)
		sle_count = frappe.db.count(
			"Stock Ledger Entry",
			{"voucher_type": "Stock Reconciliation", "voucher_no": sr.name, "is_cancelled": 0},
		)
		self.assertEqual(sle_count, 2)
		sle = frappe.get_all(
			"Stock Ledger Entry",
			{"voucher_type": "Stock Reconciliation", "voucher_no": sr.name, "is_cancelled": 0},
			["item_code", "id_plant", "actual_qty", "valuation_rate"],
		)
		for s in sle:
			if s.id_plant == plant_a.name:
				self.assertEqual(s.actual_qty, 5)
			elif s.id_plant == plant_b.name:
				self.assertEqual(s.actual_qty, 3)

	def test_serial_no_status_with_backdated_stock_reco(self):
		from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note

		item_code = self.make_item(
			"Test Item",
			{
				"is_stock_item": 1,
				"has_serial_no": 1,
				"serial_no_series": "SERIAL.###",
			},
		).name

		warehouse = "_Test Warehouse - _TC"

		reco = create_stock_reconciliation(
			item_code=item_code,
			posting_date=add_days(nowdate(), -2),
			warehouse=warehouse,
			qty=1,
			rate=80,
			purpose="Opening Stock",
		)

		serial_no = get_serial_nos_from_bundle(reco.items[0].serial_and_batch_bundle)[0]

		create_delivery_note(
			item_code=item_code, warehouse=warehouse, qty=1, rate=100, posting_date=nowdate()
		)

		self.assertEqual(frappe.get_value("Serial No", serial_no, "status"), "Delivered")

		reco = create_stock_reconciliation(
			item_code=item_code,
			posting_date=add_days(nowdate(), -1),
			warehouse=warehouse,
			qty=1,
			rate=90,
		)

		self.assertEqual(frappe.get_value("Serial No", serial_no, "status"), "Delivered")


def create_batch_item_with_batch(item_name, batch_id):
	batch_item_doc = create_item(item_name, is_stock_item=1)
	if not batch_item_doc.has_batch_no:
		batch_item_doc.has_batch_no = 1
		batch_item_doc.create_new_batch = 1
		batch_item_doc.save(ignore_permissions=True)

	if not frappe.db.exists("Batch", batch_id):
		b = frappe.new_doc("Batch")
		b.item = item_name
		b.batch_id = batch_id
		b.save()


def insert_existing_sle(warehouse, item_code="_Test Item"):
	from erpnext.stock.doctype.stock_entry.test_stock_entry import make_stock_entry

	se1 = make_stock_entry(
		posting_date="2012-12-15",
		posting_time="02:00",
		item_code=item_code,
		target=warehouse,
		qty=10,
		basic_rate=700,
	)

	se2 = make_stock_entry(
		posting_date="2012-12-25", posting_time="03:00", item_code=item_code, source=warehouse, qty=15
	)

	se3 = make_stock_entry(
		posting_date="2013-01-05",
		posting_time="07:00",
		item_code=item_code,
		target=warehouse,
		qty=15,
		basic_rate=1200,
	)

	return se1, se2, se3


def create_stock_reconciliation(**args):
	args = frappe._dict(args)
	sr = frappe.new_doc("Stock Reconciliation")
	sr.purpose = args.purpose or "Stock Reconciliation"
	sr.posting_date = args.posting_date or nowdate()
	sr.posting_time = args.posting_time or nowtime()
	sr.set_posting_time = 1
	sr.company = args.company or "_Test Company"
	sr.expense_account = args.expense_account or (
		(
			frappe.get_cached_value("Company", sr.company, "stock_adjustment_account")
			or frappe.get_cached_value(
				"Account", {"account_type": "Stock Adjustment", "company": sr.company}, "name"
			)
		)
		if frappe.get_all("Stock Ledger Entry", {"company": sr.company})
		else frappe.get_cached_value("Account", {"account_type": "Temporary", "company": sr.company}, "name")
	)
	sr.cost_center = (
		args.cost_center
		or frappe.get_cached_value("Company", sr.company, "cost_center")
		or frappe.get_cached_value("Cost Center", filters={"is_group": 0, "company": sr.company})
	)

	bundle_id = None
	if not args.use_serial_batch_fields and (args.batch_no or args.serial_no) and args.qty:
		batches = frappe._dict({})
		if args.batch_no:
			batches[args.batch_no] = args.qty

		bundle_id = make_serial_batch_bundle(
			frappe._dict(
				{
					"item_code": args.item_code or "_Test Item",
					"warehouse": args.warehouse or "_Test Warehouse - _TC",
					"qty": args.qty,
					"voucher_type": "Stock Reconciliation",
					"batches": batches,
					"rate": args.rate,
					"serial_nos": args.serial_no,
					"posting_date": sr.posting_date,
					"posting_time": sr.posting_time,
					"type_of_transaction": "Inward" if args.qty > 0 else "Outward",
					"company": args.company or "_Test Company",
					"do_not_submit": True,
				}
			)
		).name

	if args.reconcile_all_serial_batch is None:
		args.reconcile_all_serial_batch = 1

	sr.append(
		"items",
		{
			"item_code": args.item_code or "_Test Item",
			"warehouse": args.warehouse or "_Test Warehouse - _TC",
			"qty": args.qty,
			"reconcile_all_serial_batch": args.reconcile_all_serial_batch,
			"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,
		},
	)

	if not args.do_not_save:
		sr.insert()
		try:
			if not args.do_not_submit:
				sr.submit()
		except EmptyStockReconciliationItemsError:
			pass

		sr.load_from_db()

	return sr


def set_valuation_method(item_code, valuation_method):
	existing_valuation_method = get_valuation_method(item_code)
	if valuation_method == existing_valuation_method:
		return

	frappe.db.set_value("Item", item_code, "valuation_method", valuation_method)

	for warehouse in frappe.get_all(
		"Warehouse", filters={"company": "_Test Company"}, fields=["name", "is_group"]
	):
		if not warehouse.is_group:
			update_entries_after(
				{"item_code": item_code, "warehouse": warehouse.name}, allow_negative_stock=1
			)
