# Copyright (c) 2020, Frappe Technologies Pvt. Ltd. and Contributors
# See license.txt

import copy
import unittest

import frappe
from frappe import _
from frappe.tests.utils import FrappeTestCase
from frappe.utils import add_days, nowdate

from erpnext.accounts.doctype.mode_of_payment.test_mode_of_payment import (
	set_default_account_for_mode_of_payment,
)
from erpnext.accounts.doctype.pos_invoice.pos_invoice import (
	PartialPaymentValidationError,
	make_merge_log,
	make_sales_return,
)
from erpnext.accounts.doctype.pos_profile.test_pos_profile import make_pos_profile
from erpnext.accounts.doctype.sales_invoice.test_sales_invoice import create_sales_invoice
from erpnext.stock.doctype.item.test_item import make_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_entry.stock_entry_utils import make_stock_entry
from erpnext.tests.permission_test_utils import (
	OTHER_COMPANY,
	as_user,
	assert_refused,
	assert_refused_for_names,
	assert_refused_without,
	insert_test_record,
	make_company_fenced_user,
	make_fenced_user,
)


class TestPOSInvoice(unittest.TestCase):
	@classmethod
	def setUpClass(cls):
		make_stock_entry(target="_Test Warehouse - _TC", item_code="_Test Item", qty=800, basic_rate=100)
		frappe.db.sql("delete from `tabTax Rule`")

		from erpnext.accounts.doctype.pos_closing_entry.test_pos_closing_entry import init_user_and_profile
		from erpnext.accounts.doctype.pos_opening_entry.test_pos_opening_entry import create_opening_entry

		cls.test_user, cls.pos_profile = init_user_and_profile()
		create_opening_entry(cls.pos_profile, cls.test_user)
		mode_of_payment = frappe.get_doc("Mode of Payment", "Bank Draft")
		set_default_account_for_mode_of_payment(mode_of_payment, "_Test Company", "_Test Bank - _TC")

	def tearDown(self):
		if frappe.session.user != "Administrator":
			frappe.set_user("Administrator")

		if frappe.db.get_single_value("Selling Settings", "validate_selling_price"):
			frappe.db.set_single_value("Selling Settings", "validate_selling_price", 0)

	def test_timestamp_change(self):
		w = create_pos_invoice(do_not_save=1)
		w.docstatus = 0
		w.insert()

		w2 = frappe.get_doc(w.doctype, w.name)

		import time

		time.sleep(1)
		w.save()

		import time

		time.sleep(1)
		self.assertRaises(frappe.TimestampMismatchError, w2.save)

	def test_change_naming_series(self):
		inv = create_pos_invoice(do_not_submit=1)
		inv.naming_series = "TEST-"

		self.assertRaises(frappe.CannotChangeConstantError, inv.save)

	def test_discount_and_inclusive_tax(self):
		inv = create_pos_invoice(qty=100, rate=50, do_not_save=1)
		inv.append(
			"taxes",
			{
				"charge_type": "On Net Total",
				"account_head": "_Test Account Service Tax - _TC",
				"cost_center": "_Test Cost Center - _TC",
				"description": "Service Tax",
				"rate": 14,
				"included_in_print_rate": 1,
			},
		)
		inv.insert()

		self.assertEqual(inv.net_total, 4385.96)
		self.assertEqual(inv.grand_total, 5000)

		inv.reload()

		inv.discount_amount = 100
		inv.apply_discount_on = "Net Total"
		inv.payment_schedule = []

		inv.save()

		self.assertEqual(inv.net_total, 4285.96)
		self.assertEqual(inv.grand_total, 4885.99)

		inv.reload()

		inv.discount_amount = 100
		inv.apply_discount_on = "Grand Total"
		inv.payment_schedule = []

		inv.save()

		self.assertEqual(inv.net_total, 4298.24)
		self.assertEqual(inv.grand_total, 4900.00)

	def test_tax_calculation_with_multiple_items(self):
		inv = create_pos_invoice(qty=84, rate=4.6, do_not_save=True)
		item_row = inv.get("items")[0]
		for qty in (54, 288, 144, 430):
			item_row_copy = copy.deepcopy(item_row)
			item_row_copy.qty = qty
			inv.append("items", item_row_copy)

		inv.append(
			"taxes",
			{
				"account_head": "_Test Account VAT - _TC",
				"charge_type": "On Net Total",
				"cost_center": "_Test Cost Center - _TC",
				"description": "VAT",
				"doctype": "Sales Taxes and Charges",
				"rate": 19,
			},
		)
		inv.insert()

		self.assertEqual(inv.net_total, 4600)

		self.assertEqual(inv.get("taxes")[0].tax_amount, 874.0)
		self.assertEqual(inv.get("taxes")[0].total, 5474.0)

		self.assertEqual(inv.grand_total, 5474.0)

	def test_tax_calculation_with_item_tax_template(self):
		inv = create_pos_invoice(qty=84, rate=4.6, do_not_save=1)
		item_row = inv.get("items")[0]

		add_items = [
			(54, "_Test Account Excise Duty @ 12 - _TC"),
			(288, "_Test Account Excise Duty @ 15 - _TC"),
			(144, "_Test Account Excise Duty @ 20 - _TC"),
			(430, "_Test Item Tax Template 1 - _TC"),
		]
		for qty, item_tax_template in add_items:
			item_row_copy = copy.deepcopy(item_row)
			item_row_copy.qty = qty
			item_row_copy.item_tax_template = item_tax_template
			inv.append("items", item_row_copy)

		inv.append(
			"taxes",
			{
				"account_head": "_Test Account Excise Duty - _TC",
				"charge_type": "On Net Total",
				"cost_center": "_Test Cost Center - _TC",
				"description": "Excise Duty",
				"doctype": "Sales Taxes and Charges",
				"rate": 11,
			},
		)
		inv.append(
			"taxes",
			{
				"account_head": "_Test Account Education Cess - _TC",
				"charge_type": "On Net Total",
				"cost_center": "_Test Cost Center - _TC",
				"description": "Education Cess",
				"doctype": "Sales Taxes and Charges",
				"rate": 0,
			},
		)
		inv.append(
			"taxes",
			{
				"account_head": "_Test Account S&H Education Cess - _TC",
				"charge_type": "On Net Total",
				"cost_center": "_Test Cost Center - _TC",
				"description": "S&H Education Cess",
				"doctype": "Sales Taxes and Charges",
				"rate": 3,
			},
		)
		inv.insert()

		self.assertEqual(inv.net_total, 4600)

		self.assertEqual(inv.get("taxes")[0].tax_amount, 502.41)
		self.assertEqual(inv.get("taxes")[0].total, 5102.41)

		self.assertEqual(inv.get("taxes")[1].tax_amount, 197.80)
		self.assertEqual(inv.get("taxes")[1].total, 5300.21)

		self.assertEqual(inv.get("taxes")[2].tax_amount, 375.36)
		self.assertEqual(inv.get("taxes")[2].total, 5675.57)

		self.assertEqual(inv.grand_total, 5675.57)
		self.assertEqual(inv.rounding_adjustment, 0.43)
		self.assertEqual(inv.rounded_total, 5676.0)

	def test_tax_calculation_with_multiple_items_and_discount(self):
		inv = create_pos_invoice(qty=1, rate=75, do_not_save=True)
		item_row = inv.get("items")[0]
		for rate in (500, 200, 100, 50, 50):
			item_row_copy = copy.deepcopy(item_row)
			item_row_copy.price_list_rate = rate
			item_row_copy.rate = rate
			inv.append("items", item_row_copy)

		inv.apply_discount_on = "Net Total"
		inv.discount_amount = 75.0

		inv.append(
			"taxes",
			{
				"account_head": "_Test Account VAT - _TC",
				"charge_type": "On Net Total",
				"cost_center": "_Test Cost Center - _TC",
				"description": "VAT",
				"doctype": "Sales Taxes and Charges",
				"rate": 24,
			},
		)
		inv.insert()

		self.assertEqual(inv.total, 975)
		self.assertEqual(inv.net_total, 900)

		self.assertEqual(inv.get("taxes")[0].tax_amount, 216.0)
		self.assertEqual(inv.get("taxes")[0].total, 1116.0)

		self.assertEqual(inv.grand_total, 1116.0)

	def test_pos_returns_with_repayment(self):
		pos = create_pos_invoice(qty=10, do_not_save=True)

		pos.set("payments", [])
		pos.append("payments", {"mode_of_payment": "Bank Draft", "amount": 500})
		pos.append("payments", {"mode_of_payment": "Cash", "amount": 500, "default": 1})
		pos.insert()
		pos.submit()

		pos_return = make_sales_return(pos.name)

		pos_return.insert()
		pos_return.submit()

		self.assertEqual(pos_return.get("payments")[0].amount, -500)
		self.assertEqual(pos_return.get("payments")[1].amount, -500)

	def test_pos_return_for_serialized_item(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_serialized_item

		se = make_serialized_item(
			company="_Test Company",
			target_warehouse="Stores - _TC",
			cost_center="Main - _TC",
			expense_account="Cost of Goods Sold - _TC",
		)

		serial_nos = get_serial_nos_from_bundle(se.get("items")[0].serial_and_batch_bundle)

		pos = create_pos_invoice(
			company="_Test Company",
			debit_to="Debtors - _TC",
			account_for_change_amount="Cash - _TC",
			warehouse="Stores - _TC",
			income_account="Sales - _TC",
			expense_account="Cost of Goods Sold - _TC",
			cost_center="Main - _TC",
			item=se.get("items")[0].item_code,
			serial_no=[serial_nos[0]],
			rate=1000,
			do_not_save=1,
		)

		pos.append("payments", {"mode_of_payment": "Cash", "amount": 1000, "default": 1})

		pos.insert()
		pos.submit()

		pos_return = make_sales_return(pos.name)

		pos_return.insert()
		pos_return.submit()
		self.assertEqual(
			get_serial_nos_from_bundle(pos_return.get("items")[0].serial_and_batch_bundle)[0], serial_nos[0]
		)

	def test_partial_pos_returns(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_serialized_item

		se = make_serialized_item(
			company="_Test Company",
			target_warehouse="Stores - _TC",
			cost_center="Main - _TC",
			expense_account="Cost of Goods Sold - _TC",
		)

		serial_nos = get_serial_nos_from_bundle(se.get("items")[0].serial_and_batch_bundle)

		pos = create_pos_invoice(
			company="_Test Company",
			debit_to="Debtors - _TC",
			account_for_change_amount="Cash - _TC",
			warehouse="Stores - _TC",
			income_account="Sales - _TC",
			expense_account="Cost of Goods Sold - _TC",
			cost_center="Main - _TC",
			item=se.get("items")[0].item_code,
			serial_no=serial_nos,
			qty=2,
			rate=1000,
			do_not_save=1,
		)

		pos.append("payments", {"mode_of_payment": "Cash", "amount": 2000, "default": 1})

		pos.insert()
		pos.submit()
		pos.reload()

		pos_return1 = make_sales_return(pos.name)

		# partial return 1
		pos_return1.get("items")[0].qty = -1
		pos_return1.set("payments", [])
		pos_return1.append("payments", {"mode_of_payment": "Cash", "amount": -1000, "default": 1})
		pos_return1.paid_amount = -1000
		pos_return1.submit()
		pos_return1.reload()

		bundle_id = frappe.get_doc(
			"Serial and Batch Bundle", pos_return1.get("items")[0].serial_and_batch_bundle
		)

		bundle_id.load_from_db()

		serial_no = bundle_id.entries[0].serial_no
		self.assertEqual(serial_no, serial_nos[0])

		# partial return 2
		pos_return2 = make_sales_return(pos.name)
		pos_return2.set("payments", [])
		pos_return2.append("payments", {"mode_of_payment": "Cash", "amount": -1000, "default": 1})
		pos_return2.paid_amount = -1000
		pos_return2.submit()

		self.assertEqual(pos_return2.get("items")[0].qty, -1)
		serial_no = get_serial_nos_from_bundle(pos_return2.get("items")[0].serial_and_batch_bundle)[0]
		self.assertEqual(serial_no, serial_nos[1])

	def test_pos_change_amount(self):
		pos = create_pos_invoice(
			company="_Test Company",
			debit_to="Debtors - _TC",
			income_account="Sales - _TC",
			expense_account="Cost of Goods Sold - _TC",
			rate=105,
			cost_center="Main - _TC",
			do_not_save=True,
		)

		pos.set("payments", [])
		pos.append("payments", {"mode_of_payment": "Bank Draft", "amount": 50})
		pos.append("payments", {"mode_of_payment": "Cash", "amount": 60, "default": 1})

		pos.insert()
		pos.submit()

		self.assertEqual(pos.grand_total, 105.0)
		self.assertEqual(pos.change_amount, 5.0)

	def test_without_payment(self):
		inv = create_pos_invoice(do_not_save=1)
		# Check that the invoice cannot be submitted without payments
		inv.payments = []
		self.assertRaises(frappe.ValidationError, inv.insert)

	def test_partial_payment(self):
		pos_inv = create_pos_invoice(rate=10000, do_not_save=1)
		pos_inv.append(
			"payments",
			{"mode_of_payment": "Cash", "amount": 9000},
		)
		pos_inv.insert()
		self.assertRaises(PartialPaymentValidationError, pos_inv.submit)

	def test_discounted_invoice_status(self):
		allow_partial_payment = self.pos_profile.allow_partial_payment
		self.pos_profile.db_set("allow_partial_payment", 1)
		self.addCleanup(self.pos_profile.db_set, "allow_partial_payment", allow_partial_payment)
		# v15 does not calculate outstanding_amount for POS Invoice.
		pos_inv = create_pos_invoice(
			pos_profile=self.pos_profile.name,
			rate=100,
			is_discounted=1,
			outstanding_amount=10,
			do_not_save=1,
		)
		pos_inv.append("payments", {"mode_of_payment": "Cash", "amount": 90})
		pos_inv.insert()

		# Seed the lookup records because Invoice Discounting only accepts Sales Invoice links.
		discounting = frappe.get_doc(
			doctype="Invoice Discounting",
			name=frappe.generate_hash(length=10),
			company=pos_inv.company,
			docstatus=1,
			status="Disbursed",
		)
		discounting.db_insert()
		frappe.get_doc(
			doctype="Discounted Invoice",
			parent=discounting.name,
			parenttype=discounting.doctype,
			parentfield="invoices",
			sales_invoice=pos_inv.name,
			docstatus=1,
		).db_insert()

		pos_inv.submit()
		pos_inv.reload()
		self.assertEqual(pos_inv.docstatus, 1)
		self.assertEqual(pos_inv.outstanding_amount, 10)
		self.assertEqual(pos_inv.status, "Unpaid and Discounted")

		for outstanding_amount, due_date, expected_status in (
			(10, add_days(nowdate(), -1), "Overdue"),
			(10, nowdate(), "Unpaid"),
			(100, nowdate(), "Unpaid"),
		):
			with self.subTest(status=expected_status):
				pos_inv.outstanding_amount = outstanding_amount
				pos_inv.due_date = due_date
				discounting.db_set("status", "Disbursed")
				pos_inv.set_status()
				self.assertEqual(pos_inv.status, f"{expected_status} and Discounted")

				discounting.db_set("status", "Settled")
				pos_inv.set_status()
				self.assertEqual(pos_inv.status, expected_status)

	def test_serialized_item_transaction(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_serialized_item

		se = make_serialized_item(
			company="_Test Company",
			target_warehouse="Stores - _TC",
			cost_center="Main - _TC",
			expense_account="Cost of Goods Sold - _TC",
		)

		serial_nos = get_serial_nos_from_bundle(se.get("items")[0].serial_and_batch_bundle)

		pos = create_pos_invoice(
			company="_Test Company",
			debit_to="Debtors - _TC",
			account_for_change_amount="Cash - _TC",
			warehouse="Stores - _TC",
			income_account="Sales - _TC",
			expense_account="Cost of Goods Sold - _TC",
			cost_center="Main - _TC",
			item=se.get("items")[0].item_code,
			rate=1000,
			serial_no=[serial_nos[0]],
			do_not_save=1,
		)

		pos.append("payments", {"mode_of_payment": "Bank Draft", "amount": 1000})

		pos.insert()
		pos.submit()

		pos2 = create_pos_invoice(
			company="_Test Company",
			debit_to="Debtors - _TC",
			account_for_change_amount="Cash - _TC",
			warehouse="Stores - _TC",
			income_account="Sales - _TC",
			expense_account="Cost of Goods Sold - _TC",
			cost_center="Main - _TC",
			item=se.get("items")[0].item_code,
			rate=1000,
			serial_no=[serial_nos[0]],
			do_not_save=1,
		)

		pos2.append("payments", {"mode_of_payment": "Bank Draft", "amount": 1000})

		pos2.insert()
		self.assertRaises(frappe.ValidationError, pos2.submit)

	def test_delivered_serialized_item_transaction(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_serialized_item

		se = make_serialized_item(
			company="_Test Company",
			target_warehouse="Stores - _TC",
			cost_center="Main - _TC",
			expense_account="Cost of Goods Sold - _TC",
		)

		serial_nos = get_serial_nos_from_bundle(se.get("items")[0].serial_and_batch_bundle)

		si = create_sales_invoice(
			company="_Test Company",
			debit_to="Debtors - _TC",
			account_for_change_amount="Cash - _TC",
			warehouse="Stores - _TC",
			income_account="Sales - _TC",
			expense_account="Cost of Goods Sold - _TC",
			cost_center="Main - _TC",
			item=se.get("items")[0].item_code,
			rate=1000,
			update_stock=1,
			serial_no=[serial_nos[0]],
			do_not_save=1,
		)

		si.insert()
		si.submit()

		pos2 = create_pos_invoice(
			company="_Test Company",
			debit_to="Debtors - _TC",
			account_for_change_amount="Cash - _TC",
			warehouse="Stores - _TC",
			income_account="Sales - _TC",
			expense_account="Cost of Goods Sold - _TC",
			cost_center="Main - _TC",
			item=se.get("items")[0].item_code,
			rate=1000,
			serial_no=[serial_nos[0]],
			do_not_save=1,
			ignore_sabb_validation=True,
		)

		pos2.append("payments", {"mode_of_payment": "Bank Draft", "amount": 1000})

		pos2.insert()
		self.assertRaises(frappe.ValidationError, pos2.submit)

	def test_invalid_serial_no_validation(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_serialized_item

		se = make_serialized_item(
			company="_Test Company",
			target_warehouse="Stores - _TC",
			cost_center="Main - _TC",
			expense_account="Cost of Goods Sold - _TC",
		)
		serial_nos = get_serial_nos_from_bundle(se.get("items")[0].serial_and_batch_bundle)[0] + "wrong"

		pos = create_pos_invoice(
			company="_Test Company",
			debit_to="Debtors - _TC",
			account_for_change_amount="Cash - _TC",
			warehouse="Stores - _TC",
			income_account="Sales - _TC",
			expense_account="Cost of Goods Sold - _TC",
			cost_center="Main - _TC",
			item=se.get("items")[0].item_code,
			rate=1000,
			qty=2,
			serial_nos=[serial_nos],
			do_not_save=1,
		)

		pos.get("items")[0].has_serial_no = 1

		self.assertRaises(frappe.ValidationError, pos.insert)

	def test_value_error_on_serial_no_validation(self):
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_serialized_item

		se = make_serialized_item(
			company="_Test Company",
			target_warehouse="Stores - _TC",
			cost_center="Main - _TC",
			expense_account="Cost of Goods Sold - _TC",
		)
		serial_nos = get_serial_nos_from_bundle(se.get("items")[0].serial_and_batch_bundle)

		# make a pos invoice
		pos = create_pos_invoice(
			company="_Test Company",
			debit_to="Debtors - _TC",
			account_for_change_amount="Cash - _TC",
			warehouse="Stores - _TC",
			income_account="Sales - _TC",
			expense_account="Cost of Goods Sold - _TC",
			cost_center="Main - _TC",
			item=se.get("items")[0].item_code,
			rate=1000,
			serial_no=[serial_nos[0]],
			qty=1,
			do_not_save=1,
		)
		pos.get("items")[0].has_serial_no = 1
		pos.set("payments", [])
		pos.append("payments", {"mode_of_payment": "Cash", "amount": 1000, "default": 1})
		pos = pos.save().submit()

		# make a return
		pos_return = make_sales_return(pos.name)
		pos_return.paid_amount = pos_return.grand_total
		pos_return.save()
		pos_return.submit()

		# set docstatus to 2 for pos to trigger this issue
		frappe.db.set_value("POS Invoice", pos.name, "docstatus", 2)

		pos2 = create_pos_invoice(
			company="_Test Company",
			debit_to="Debtors - _TC",
			account_for_change_amount="Cash - _TC",
			warehouse="Stores - _TC",
			income_account="Sales - _TC",
			expense_account="Cost of Goods Sold - _TC",
			cost_center="Main - _TC",
			item=se.get("items")[0].item_code,
			rate=1000,
			serial_no=[serial_nos[0]],
			qty=1,
			do_not_save=1,
		)

		pos2.get("items")[0].has_serial_no = 1
		# Value error should not be triggered on validation
		pos2.save()

	def test_loyalty_points(self):
		from erpnext.accounts.doctype.loyalty_program.loyalty_program import (
			get_loyalty_program_details_with_points,
		)
		from erpnext.accounts.doctype.loyalty_program.test_loyalty_program import create_records

		create_records()
		frappe.db.set_value("Customer", "Test Loyalty Customer", "loyalty_program", "Test Single Loyalty")
		before_lp_details = get_loyalty_program_details_with_points(
			"Test Loyalty Customer", company="_Test Company", loyalty_program="Test Single Loyalty"
		)

		inv = create_pos_invoice(customer="Test Loyalty Customer", rate=10000, do_not_save=1)
		inv.append(
			"payments",
			{"mode_of_payment": "Cash", "amount": 10000},
		)
		inv.insert()
		inv.submit()

		lpe = frappe.get_doc(
			"Loyalty Point Entry",
			{"invoice_type": "POS Invoice", "invoice": inv.name, "customer": inv.customer},
		)
		after_lp_details = get_loyalty_program_details_with_points(
			inv.customer, company=inv.company, loyalty_program=inv.loyalty_program
		)

		self.assertEqual(inv.get("loyalty_program"), "Test Single Loyalty")
		self.assertEqual(lpe.loyalty_points, 10)
		self.assertEqual(after_lp_details.loyalty_points, before_lp_details.loyalty_points + 10)

		inv.cancel()
		after_cancel_lp_details = get_loyalty_program_details_with_points(
			inv.customer, company=inv.company, loyalty_program=inv.loyalty_program
		)
		self.assertEqual(after_cancel_lp_details.loyalty_points, before_lp_details.loyalty_points)

	def test_loyalty_points_redeemption(self):
		from erpnext.accounts.doctype.loyalty_program.loyalty_program import (
			get_loyalty_program_details_with_points,
		)

		# add 10 loyalty points
		pos_inv = create_pos_invoice(customer="Test Loyalty Customer", rate=10000, do_not_save=1)
		pos_inv.append(
			"payments",
			{"mode_of_payment": "Cash", "amount": 10000},
		)
		pos_inv.paid_amount = 10000
		pos_inv.submit()

		before_lp_details = get_loyalty_program_details_with_points(
			"Test Loyalty Customer", company="_Test Company", loyalty_program="Test Single Loyalty"
		)

		inv = create_pos_invoice(customer="Test Loyalty Customer", rate=10000, do_not_save=1)
		inv.redeem_loyalty_points = 1
		inv.loyalty_points = before_lp_details.loyalty_points
		inv.loyalty_amount = inv.loyalty_points * before_lp_details.conversion_factor
		inv.append(
			"payments",
			{"mode_of_payment": "Cash", "amount": 10000 - inv.loyalty_amount},
		)
		inv.paid_amount = 10000
		inv.submit()

		after_redeem_lp_details = get_loyalty_program_details_with_points(
			inv.customer, company=inv.company, loyalty_program=inv.loyalty_program
		)
		self.assertEqual(after_redeem_lp_details.loyalty_points, 9)

	def test_merging_into_sales_invoice_with_discount(self):
		from erpnext.accounts.doctype.pos_closing_entry.test_pos_closing_entry import (
			init_user_and_profile,
		)
		from erpnext.accounts.doctype.pos_invoice_merge_log.pos_invoice_merge_log import (
			consolidate_pos_invoices,
		)

		frappe.db.sql("delete from `tabPOS Invoice`")
		test_user, pos_profile = init_user_and_profile()
		pos_inv = create_pos_invoice(rate=300, additional_discount_percentage=10, do_not_submit=1)
		pos_inv.append("payments", {"mode_of_payment": "Cash", "amount": 270})
		pos_inv.save()
		pos_inv.submit()

		pos_inv2 = create_pos_invoice(rate=3200, do_not_submit=1)
		pos_inv2.append("payments", {"mode_of_payment": "Cash", "amount": 3200})
		pos_inv2.save()
		pos_inv2.submit()

		consolidate_pos_invoices()

		pos_inv.load_from_db()
		rounded_total = frappe.db.get_value("Sales Invoice", pos_inv.consolidated_invoice, "rounded_total")
		self.assertEqual(rounded_total, 3470)

	def test_merging_into_sales_invoice_with_discount_and_inclusive_tax(self):
		from erpnext.accounts.doctype.pos_closing_entry.test_pos_closing_entry import (
			init_user_and_profile,
		)
		from erpnext.accounts.doctype.pos_invoice_merge_log.pos_invoice_merge_log import (
			consolidate_pos_invoices,
		)

		frappe.db.sql("delete from `tabPOS Invoice`")
		test_user, pos_profile = init_user_and_profile()
		pos_inv = create_pos_invoice(rate=300, do_not_submit=1)
		pos_inv.append("payments", {"mode_of_payment": "Cash", "amount": 300})
		pos_inv.append(
			"taxes",
			{
				"charge_type": "On Net Total",
				"account_head": "_Test Account Service Tax - _TC",
				"cost_center": "_Test Cost Center - _TC",
				"description": "Service Tax",
				"rate": 14,
				"included_in_print_rate": 1,
			},
		)
		pos_inv.save()
		pos_inv.submit()

		pos_inv2 = create_pos_invoice(rate=300, qty=2, do_not_submit=1)
		pos_inv2.additional_discount_percentage = 10
		pos_inv2.append("payments", {"mode_of_payment": "Cash", "amount": 540})
		pos_inv2.append(
			"taxes",
			{
				"charge_type": "On Net Total",
				"account_head": "_Test Account Service Tax - _TC",
				"cost_center": "_Test Cost Center - _TC",
				"description": "Service Tax",
				"rate": 14,
				"included_in_print_rate": 1,
			},
		)
		pos_inv2.save()
		pos_inv2.submit()

		consolidate_pos_invoices()

		pos_inv.load_from_db()
		rounded_total = frappe.db.get_value("Sales Invoice", pos_inv.consolidated_invoice, "rounded_total")
		self.assertEqual(rounded_total, 840)

	def test_merging_with_validate_selling_price(self):
		from erpnext.accounts.doctype.pos_closing_entry.test_pos_closing_entry import (
			init_user_and_profile,
		)
		from erpnext.accounts.doctype.pos_invoice_merge_log.pos_invoice_merge_log import (
			consolidate_pos_invoices,
		)

		if not frappe.db.get_single_value("Selling Settings", "validate_selling_price"):
			frappe.db.set_single_value("Selling Settings", "validate_selling_price", 1)

		item = "Test Selling Price Validation"
		make_item(item, {"is_stock_item": 1})
		make_purchase_receipt(item_code=item, warehouse="_Test Warehouse - _TC", qty=1, rate=300)
		frappe.db.sql("delete from `tabPOS Invoice`")
		test_user, pos_profile = init_user_and_profile()
		pos_inv = create_pos_invoice(item=item, rate=300, do_not_submit=1)
		pos_inv.append("payments", {"mode_of_payment": "Cash", "amount": 300})
		pos_inv.append(
			"taxes",
			{
				"charge_type": "On Net Total",
				"account_head": "_Test Account Service Tax - _TC",
				"cost_center": "_Test Cost Center - _TC",
				"description": "Service Tax",
				"rate": 14,
				"included_in_print_rate": 1,
			},
		)
		self.assertRaises(frappe.ValidationError, pos_inv.submit)

		pos_inv2 = create_pos_invoice(item=item, rate=400, do_not_submit=1)
		pos_inv2.append("payments", {"mode_of_payment": "Cash", "amount": 400})
		pos_inv2.append(
			"taxes",
			{
				"charge_type": "On Net Total",
				"account_head": "_Test Account Service Tax - _TC",
				"cost_center": "_Test Cost Center - _TC",
				"description": "Service Tax",
				"rate": 14,
				"included_in_print_rate": 1,
			},
		)
		pos_inv2.save()
		pos_inv2.submit()

		consolidate_pos_invoices()

		pos_inv2.load_from_db()
		rounded_total = frappe.db.get_value("Sales Invoice", pos_inv2.consolidated_invoice, "rounded_total")
		self.assertEqual(rounded_total, 400)

	def test_pos_batch_reservation(self):
		from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
			get_auto_batch_nos,
		)
		from erpnext.stock.doctype.stock_reconciliation.test_stock_reconciliation import (
			create_batch_item_with_batch,
		)

		create_batch_item_with_batch("_BATCH ITEM Test For Reserve", "TestBatch-RS 02")
		se = make_stock_entry(
			target="_Test Warehouse - _TC",
			item_code="_BATCH ITEM Test For Reserve",
			qty=30,
			basic_rate=100,
		)

		se.reload()

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

		# POS Invoice 1, for the batch without bundle
		pos_inv1 = create_pos_invoice(item="_BATCH ITEM Test For Reserve", rate=300, qty=15, do_not_save=1)
		pos_inv1.append(
			"payments",
			{"mode_of_payment": "Cash", "amount": 4500},
		)
		pos_inv1.items[0].batch_no = batch_no
		pos_inv1.save()
		pos_inv1.submit()
		pos_inv1.reload()

		batches = get_auto_batch_nos(
			frappe._dict({"item_code": "_BATCH ITEM Test For Reserve", "warehouse": "_Test Warehouse - _TC"})
		)

		for batch in batches:
			if batch.batch_no == batch_no and batch.warehouse == "_Test Warehouse - _TC":
				self.assertEqual(batch.qty, 15)

		# POS Invoice 2, for the batch with bundle
		pos_inv2 = create_pos_invoice(
			item="_BATCH ITEM Test For Reserve", rate=300, qty=10, batch_no=batch_no, do_not_save=1
		)
		pos_inv2.append(
			"payments",
			{"mode_of_payment": "Cash", "amount": 3000},
		)
		pos_inv2.save()
		pos_inv2.submit()
		pos_inv2.reload()
		self.assertTrue(pos_inv2.items[0].serial_and_batch_bundle)

		batches = get_auto_batch_nos(
			frappe._dict({"item_code": "_BATCH ITEM Test For Reserve", "warehouse": "_Test Warehouse - _TC"})
		)

		for batch in batches:
			if batch.batch_no == batch_no and batch.warehouse == "_Test Warehouse - _TC":
				self.assertEqual(batch.qty, 5)

	def test_pos_batch_reservation_with_return_qty(self):
		"""
		Test POS Invoice reserved qty for batch without bundle with return invoices.
		"""
		from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
			get_auto_batch_nos,
		)
		from erpnext.stock.doctype.stock_reconciliation.test_stock_reconciliation import (
			create_batch_item_with_batch,
		)

		create_batch_item_with_batch("_Batch Item Reserve Return", "TestBatch-RR 01")
		se = make_stock_entry(
			target="_Test Warehouse - _TC",
			item_code="_Batch Item Reserve Return",
			qty=30,
			basic_rate=100,
		)

		se.reload()

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

		# POS Invoice for the batch without bundle
		pos_inv = create_pos_invoice(item="_Batch Item Reserve Return", rate=300, qty=15, do_not_save=1)
		pos_inv.append(
			"payments",
			{"mode_of_payment": "Cash", "amount": 4500},
		)
		pos_inv.items[0].batch_no = batch_no
		pos_inv.save()
		pos_inv.submit()

		# POS Invoice return
		pos_return = make_sales_return(pos_inv.name)

		pos_return.insert()
		pos_return.submit()

		batches = get_auto_batch_nos(
			frappe._dict({"item_code": "_Batch Item Reserve Return", "warehouse": "_Test Warehouse - _TC"})
		)

		for batch in batches:
			if batch.batch_no == batch_no and batch.warehouse == "_Test Warehouse - _TC":
				self.assertEqual(batch.qty, 30)

	def test_pos_batch_item_qty_validation(self):
		from erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle import (
			BatchNegativeStockError,
		)
		from erpnext.stock.doctype.stock_reconciliation.test_stock_reconciliation import (
			create_batch_item_with_batch,
		)
		from erpnext.stock.serial_batch_bundle import SerialBatchCreation

		create_batch_item_with_batch("_BATCH ITEM", "TestBatch 01")
		item = frappe.get_doc("Item", "_BATCH ITEM")

		se = make_stock_entry(
			target="_Test Warehouse - _TC",
			item_code="_BATCH ITEM",
			qty=2,
			basic_rate=100,
			batch_no="TestBatch 01",
		)

		pos_inv1 = create_pos_invoice(
			item=item.name, rate=300, qty=1, do_not_submit=1, batch_no="TestBatch 01"
		)
		pos_inv1.append(
			"payments",
			{"mode_of_payment": "Cash", "amount": 300},
		)
		pos_inv1.save()
		pos_inv1.submit()

		pos_inv2 = create_pos_invoice(item=item.name, rate=300, qty=2, do_not_submit=1)

		sn_doc = SerialBatchCreation(
			{
				"item_code": item.name,
				"warehouse": pos_inv2.items[0].warehouse,
				"voucher_type": "POS Invoice",
				"voucher_no": pos_inv2.name,
				"qty": 2,
				"avg_rate": 300,
				"batches": frappe._dict({"TestBatch 01": 2}),
				"type_of_transaction": "Outward",
				"company": pos_inv2.company,
			}
		)

		self.assertRaises(BatchNegativeStockError, sn_doc.make_serial_and_batch_bundle)

		# teardown
		pos_inv1.reload()
		pos_inv1.cancel()
		pos_inv1.delete()
		pos_inv2.reload()
		pos_inv2.delete()
		se.cancel()

	def test_ignore_pricing_rule(self):
		from erpnext.accounts.doctype.pricing_rule.test_pricing_rule import make_pricing_rule

		item_price = frappe.get_doc(
			{
				"doctype": "Item Price",
				"item_code": "_Test Item",
				"price_list": "_Test Price List",
				"price_list_rate": "450",
			}
		)
		item_price.insert()
		pr = make_pricing_rule(selling=1, priority=5, discount_percentage=10)
		pr.save()

		try:
			pos_inv = create_pos_invoice(qty=1, do_not_submit=1)
			pos_inv.items[0].rate = 300
			pos_inv.save()
			self.assertEqual(pos_inv.items[0].discount_percentage, 10)
			# rate shouldn't change
			self.assertEqual(pos_inv.items[0].rate, 405)

			pos_inv.ignore_pricing_rule = 1
			pos_inv.save()
			self.assertEqual(pos_inv.ignore_pricing_rule, 1)
			# rate should reset since pricing rules are ignored
			self.assertEqual(pos_inv.items[0].rate, 450)

			pos_inv.items[0].rate = 300
			pos_inv.save()
			self.assertEqual(pos_inv.items[0].rate, 300)

		finally:
			item_price.delete()
			pos_inv.delete()
			pr.delete()

	def test_delivered_serial_no_case(self):
		from erpnext.accounts.doctype.pos_invoice_merge_log.test_pos_invoice_merge_log import (
			init_user_and_profile,
		)
		from erpnext.stock.doctype.delivery_note.test_delivery_note import create_delivery_note
		from erpnext.stock.doctype.stock_entry.test_stock_entry import make_serialized_item

		frappe.db.savepoint("before_test_delivered_serial_no_case")
		try:
			se = make_serialized_item()
			serial_no = get_serial_nos_from_bundle(se.get("items")[0].serial_and_batch_bundle)[0]

			dn = create_delivery_note(item_code="_Test Serialized Item With Series", serial_no=[serial_no])
			delivered_serial_no = get_serial_nos_from_bundle(dn.get("items")[0].serial_and_batch_bundle)[0]

			self.assertEqual(serial_no, delivered_serial_no)

			init_user_and_profile()

			pos_inv = create_pos_invoice(
				item_code="_Test Serialized Item With Series",
				serial_no=[serial_no],
				qty=1,
				rate=100,
				do_not_submit=True,
				ignore_sabb_validation=True,
			)

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

		finally:
			frappe.db.rollback(save_point="before_test_delivered_serial_no_case")
			frappe.set_user("Administrator")

	def test_bundle_stock_availability_validation(self):
		from erpnext.accounts.doctype.pos_invoice.pos_invoice import ProductBundleStockValidationError
		from erpnext.accounts.doctype.pos_invoice_merge_log.test_pos_invoice_merge_log import (
			init_user_and_profile,
		)
		from erpnext.selling.doctype.product_bundle.test_product_bundle import make_product_bundle
		from erpnext.stock.doctype.item.test_item import create_item

		init_user_and_profile()

		frappe.set_user("Administrator")

		warehouse = "_Test Warehouse - _TC"
		company = "_Test Company"

		# Create stock sub-items
		sub_item_a = "_Test Bundle SubA"
		if not frappe.db.exists("Item", sub_item_a):
			create_item(
				item_code=sub_item_a,
				is_stock_item=1,
			)

		sub_item_b = "_Test Bundle SubB"
		if not frappe.db.exists("Item", sub_item_b):
			create_item(
				item_code=sub_item_b,
				is_stock_item=1,
			)

		# Add initial stock: SubA=5, SubB=2
		make_stock_entry(item_code=sub_item_a, target=warehouse, qty=5, company=company)
		make_stock_entry(item_code=sub_item_b, target=warehouse, qty=2, company=company)

		# Create Product Bundle: Test Bundle (SubA x2 + SubB x1)
		bundle_item = "_Test Bundle"
		if not frappe.db.exists("Item", bundle_item):
			create_item(
				item_code=bundle_item,
				is_stock_item=0,
			)

		if not frappe.db.exists("Product Bundle", bundle_item):
			make_product_bundle(parent=bundle_item, items=[sub_item_a, sub_item_b])

		# Test Case 1: Sufficient stock (bundle qty=1: requires SubA=2 (<=5), SubB=1 (<=2)) -> No error
		pos_inv_sufficient = create_pos_invoice(
			item=bundle_item,
			qty=1,
			rate=100,
			warehouse=warehouse,
			pos_profile=self.pos_profile.name,
			do_not_save=1,
		)
		pos_inv_sufficient.append("payments", {"mode_of_payment": "Cash", "amount": 100, "default": 1})
		pos_inv_sufficient.insert()
		pos_inv_sufficient.submit()

		pos_inv_sufficient.cancel()
		pos_inv_sufficient.delete()

		# Test Case 2: Insufficient stock (reduce SubB to 1, bundle qty=2: requires SubB=2 >1) -> Error with details
		make_stock_entry(item_code=sub_item_b, from_warehouse=warehouse, qty=1, company=company)

		pos_inv_insufficient = create_pos_invoice(
			item=bundle_item,
			qty=2,
			rate=100,
			warehouse=warehouse,
			pos_profile=self.pos_profile.name,
			do_not_save=1,
		)
		pos_inv_insufficient.append("payments", {"mode_of_payment": "Cash", "amount": 200, "default": 1})
		pos_inv_insufficient.save()
		self.assertRaises(ProductBundleStockValidationError, pos_inv_insufficient.submit)

		frappe.set_user("test@example.com")

	def test_serial_and_batch_item_with_only_batch_in_fields(self):
		from frappe.tests.utils import change_settings

		from erpnext.stock.doctype.serial_and_batch_bundle.test_serial_and_batch_bundle import (
			get_batch_from_bundle,
		)

		frappe.set_user("Administrator")
		item_code = make_item(
			properties={
				"is_stock_item": 1,
				"has_batch_no": 1,
				"create_new_batch": 1,
				"batch_number_series": "PSB-.#####",
				"has_serial_no": 1,
				"serial_no_series": "PSS-.#####",
			}
		).name
		se = make_stock_entry(item_code=item_code, target="_Test Warehouse - _TC", qty=5, basic_rate=100)
		batch_no = get_batch_from_bundle(se.items[0].serial_and_batch_bundle)

		with change_settings("Stock Settings", {"auto_create_serial_and_batch_bundle_for_outward": 1}):
			pos = create_pos_invoice(
				item=item_code, qty=2, rate=100, use_serial_batch_fields=1, batch_no=batch_no, do_not_save=1
			)
			pos.items[0].update({"has_serial_no": 1, "has_batch_no": 1})
			pos.append("payments", {"mode_of_payment": "Cash", "amount": 200, "default": 1})
			pos.insert()
			self.assertFalse(pos.items[0].serial_no)

			pos.submit()
			pos.reload()
			entries = frappe.get_all(
				"Serial and Batch Entry",
				{"parent": pos.items[0].serial_and_batch_bundle},
				["serial_no", "batch_no"],
			)
			self.assertEqual(len({d.serial_no for d in entries if d.serial_no}), 2)
			self.assertEqual({d.batch_no for d in entries}, {batch_no})


def create_pos_invoice(**args):
	args = frappe._dict(args)
	pos_profile = None
	if not args.pos_profile:
		pos_profile = make_pos_profile()
		pos_profile.save()

	pos_inv = frappe.new_doc("POS Invoice")
	pos_inv.update(args)
	pos_inv.is_pos = 1
	pos_inv.pos_profile = args.pos_profile or pos_profile.name

	if args.posting_date:
		pos_inv.set_posting_time = 1
	pos_inv.posting_date = args.posting_date or frappe.utils.nowdate()

	pos_inv.company = args.company or "_Test Company"
	pos_inv.customer = args.customer or "_Test Customer"
	pos_inv.debit_to = args.debit_to or "Debtors - _TC"
	pos_inv.is_return = args.is_return
	pos_inv.return_against = args.return_against
	pos_inv.currency = args.currency or "INR"
	pos_inv.conversion_rate = args.conversion_rate or 1
	pos_inv.account_for_change_amount = args.account_for_change_amount or "Cash - _TC"

	pos_inv.set_missing_values()

	bundle_id = None
	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 pos_inv.is_return:
			type_of_transaction = "Inward"

		qty = args.get("qty") or 1
		qty *= -1 if type_of_transaction == "Outward" else 1
		batches = {}
		if args.get("batch_no"):
			batches = frappe._dict({args.batch_no: qty})

		bundle_id = make_serial_batch_bundle(
			frappe._dict(
				{
					"item_code": args.item or args.item_code or "_Test Item",
					"warehouse": args.warehouse or "_Test Warehouse - _TC",
					"qty": qty,
					"batches": batches,
					"voucher_type": "Delivery Note",
					"serial_nos": args.serial_no,
					"posting_date": pos_inv.posting_date,
					"posting_time": pos_inv.posting_time,
					"type_of_transaction": type_of_transaction,
					"do_not_submit": True,
					"ignore_sabb_validation": args.ignore_sabb_validation,
				}
			)
		).name

		if not bundle_id:
			msg = f"Serial No {args.serial_no} not available for Item {args.item}"
			frappe.throw(_(msg))

	pos_invoice_item = {
		"warehouse": args.warehouse or "_Test Warehouse - _TC",
		"qty": args.qty or 1,
		"rate": args.rate if args.get("rate") is not None else 100,
		"income_account": args.income_account or "Sales - _TC",
		"expense_account": args.expense_account or "Cost of Goods Sold - _TC",
		"cost_center": args.cost_center or "_Test Cost Center - _TC",
		"serial_and_batch_bundle": bundle_id,
		"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,
	}
	# append in pos invoice items without item_code by checking flag without_item_code
	if args.without_item_code:
		pos_inv.append(
			"items",
			{
				**pos_invoice_item,
				"item_name": args.item_name or "_Test Item",
				"description": args.item_name or "_Test Item",
			},
		)

	else:
		pos_inv.append(
			"items",
			{
				**pos_invoice_item,
				"item_code": args.item or args.item_code or "_Test Item",
			},
		)

	if not args.do_not_save:
		pos_inv.insert()
		if not args.do_not_submit:
			pos_inv.submit()
			if args.use_serial_batch_fields:
				pos_inv.reload()
		else:
			pos_inv.payment_schedule = []
	else:
		pos_inv.payment_schedule = []

	return pos_inv


class TestPOSInvoiceMergeLogPermissions(FrappeTestCase):
	def setUp(self):
		self.invoice_a = self.insert_invoice("_Test Customer")
		self.invoice_b = self.insert_invoice("_Test Customer 1")

	def tearDown(self):
		frappe.db.rollback()

	def insert_invoice(self, customer):
		return insert_test_record(
			"POS Invoice",
			{
				"company": "_Test Company",
				"customer": customer,
				"posting_date": frappe.utils.nowdate(),
				"grand_total": 100,
				"docstatus": 1,
			},
		)

	def merge_log_kwargs(self, name):
		return {"invoices": frappe.as_json([{"name": name}])}

	def test_make_merge_log_refuses_an_invoice_outside_the_customer_fence(self):
		fenced = make_fenced_user(
			"pos-merge-fenced@example.com", ["Sales User"], [("Customer", "_Test Customer")]
		)
		with as_user(fenced):
			assert_refused_for_names(
				self, make_merge_log, self.merge_log_kwargs, [self.invoice_b], caller_supplied=True
			)
			merge_log = make_merge_log(**self.merge_log_kwargs(self.invoice_a))
		self.assertEqual(merge_log["pos_invoices"][0]["customer"], "_Test Customer")

	def test_make_merge_log_refuses_an_invoice_outside_the_company_fence(self):
		fenced = make_company_fenced_user("pos-merge-company@example.com", ["Sales User"], OTHER_COMPANY)
		with as_user(fenced):
			assert_refused_without(
				self,
				["linked to", "'_Test Company'"],
				make_merge_log,
				**self.merge_log_kwargs(self.invoice_a),
			)

	def test_make_merge_log_allows_an_unfenced_sales_user(self):
		user = make_fenced_user("pos-merge-open@example.com", ["Sales User"])
		with as_user(user):
			merge_log = make_merge_log(**self.merge_log_kwargs(self.invoice_b))
		self.assertEqual(merge_log["pos_invoices"][0]["grand_total"], 100)