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

import frappe
from frappe import qb
from frappe.query_builder.functions import Count, Sum
from frappe.utils import add_days, nowdate

from erpnext.accounts.doctype.payment_entry.payment_entry import get_payment_entry
from erpnext.accounts.doctype.payment_entry.test_payment_entry import create_payment_entry
from erpnext.accounts.doctype.sales_invoice.test_sales_invoice import create_sales_invoice
from erpnext.selling.doctype.sales_order.test_sales_order import make_sales_order
from erpnext.tests.utils import ERPNextTestSuite


class TestPaymentLedgerEntry(ERPNextTestSuite):
	def setUp(self):
		self.ple = qb.DocType("Payment Ledger Entry")
		self.company = "_Test Company"
		self.cost_center = "Main - _TC"
		self.warehouse = "Stores - _TC"
		self.income_account = "Sales - _TC"
		self.expense_account = "Cost of Goods Sold - _TC"
		self.debit_to = "Debtors - _TC"
		self.creditors = "Creditors - _TC"
		self.bank = "Cash - _TC"
		self.item = "_Test Item"
		self.customer = "_Test Customer"

	def create_sales_invoice(
		self, qty=1, rate=100, posting_date=None, do_not_save=False, do_not_submit=False
	):
		"""
		Helper function to populate default values in sales invoice
		"""
		if posting_date is None:
			posting_date = nowdate()

		sinv = create_sales_invoice(
			posting_date=posting_date,
			qty=qty,
			rate=rate,
			company=self.company,
			customer=self.customer,
			item_code=self.item,
			item_name=self.item,
			cost_center=self.cost_center,
			warehouse=self.warehouse,
			debit_to=self.debit_to,
			parent_cost_center=self.cost_center,
			update_stock=0,
			currency="INR",
			is_pos=0,
			is_return=0,
			return_against=None,
			income_account=self.income_account,
			expense_account=self.expense_account,
			do_not_save=do_not_save,
			do_not_submit=do_not_submit,
		)
		return sinv

	def create_payment_entry(self, amount=100, posting_date=None):
		"""
		Helper function to populate default values in payment entry
		"""
		if posting_date is None:
			posting_date = nowdate()
		payment = create_payment_entry(
			company=self.company,
			payment_type="Receive",
			party_type="Customer",
			party=self.customer,
			paid_from=self.debit_to,
			paid_to=self.bank,
			paid_amount=amount,
		)
		payment.posting_date = posting_date
		return payment

	def create_sales_order(self, qty=1, rate=100, posting_date=None, do_not_save=False, do_not_submit=False):
		if posting_date is None:
			posting_date = nowdate()

		so = make_sales_order(
			company=self.company,
			transaction_date=posting_date,
			customer=self.customer,
			item_code=self.item,
			cost_center=self.cost_center,
			warehouse=self.warehouse,
			debit_to=self.debit_to,
			currency="INR",
			qty=qty,
			rate=100,
			do_not_save=do_not_save,
			do_not_submit=do_not_submit,
		)
		return so

	def create_journal_entry(self, acc1=None, acc2=None, amount=0, posting_date=None, cost_center=None):
		je = frappe.new_doc("Journal Entry")
		je.posting_date = posting_date or nowdate()
		je.company = self.company
		je.user_remark = "test"
		if not cost_center:
			cost_center = self.cost_center
		je.set(
			"accounts",
			[
				{
					"account": acc1,
					"cost_center": cost_center,
					"debit_in_account_currency": amount if amount > 0 else 0,
					"credit_in_account_currency": abs(amount) if amount < 0 else 0,
				},
				{
					"account": acc2,
					"cost_center": cost_center,
					"credit_in_account_currency": amount if amount > 0 else 0,
					"debit_in_account_currency": abs(amount) if amount < 0 else 0,
				},
			],
		)
		return je

	def test_payment_against_invoice(self):
		transaction_date = nowdate()
		amount = 100
		ple = self.ple

		# full payment using PE
		si1 = self.create_sales_invoice(qty=1, rate=amount, posting_date=transaction_date)
		pe1 = get_payment_entry(si1.doctype, si1.name).save().submit()

		pl_entries = (
			qb.from_(ple)
			.select(
				ple.voucher_type,
				ple.voucher_no,
				ple.against_voucher_type,
				ple.against_voucher_no,
				ple.amount,
				ple.delinked,
			)
			.where((ple.against_voucher_type == si1.doctype) & (ple.against_voucher_no == si1.name))
			.orderby(ple.creation)
			.run(as_dict=True)
		)

		expected_values = [
			{
				"voucher_type": si1.doctype,
				"voucher_no": si1.name,
				"against_voucher_type": si1.doctype,
				"against_voucher_no": si1.name,
				"amount": amount,
				"delinked": 0,
			},
			{
				"voucher_type": pe1.doctype,
				"voucher_no": pe1.name,
				"against_voucher_type": si1.doctype,
				"against_voucher_no": si1.name,
				"amount": -amount,
				"delinked": 0,
			},
		]
		self.assertEqual(pl_entries[0], expected_values[0])
		self.assertEqual(pl_entries[1], expected_values[1])

	def test_partial_payment_against_invoice(self):
		ple = self.ple
		transaction_date = nowdate()
		amount = 100

		# partial payment of invoice using PE
		si2 = self.create_sales_invoice(qty=1, rate=amount, posting_date=transaction_date)
		pe2 = get_payment_entry(si2.doctype, si2.name)
		pe2.get("references")[0].allocated_amount = 50
		pe2.get("references")[0].outstanding_amount = 50
		pe2 = pe2.save().submit()

		pl_entries = (
			qb.from_(ple)
			.select(
				ple.voucher_type,
				ple.voucher_no,
				ple.against_voucher_type,
				ple.against_voucher_no,
				ple.amount,
				ple.delinked,
			)
			.where((ple.against_voucher_type == si2.doctype) & (ple.against_voucher_no == si2.name))
			.orderby(ple.creation)
			.run(as_dict=True)
		)

		expected_values = [
			{
				"voucher_type": si2.doctype,
				"voucher_no": si2.name,
				"against_voucher_type": si2.doctype,
				"against_voucher_no": si2.name,
				"amount": amount,
				"delinked": 0,
			},
			{
				"voucher_type": pe2.doctype,
				"voucher_no": pe2.name,
				"against_voucher_type": si2.doctype,
				"against_voucher_no": si2.name,
				"amount": -50,
				"delinked": 0,
			},
		]
		self.assertEqual(pl_entries[0], expected_values[0])
		self.assertEqual(pl_entries[1], expected_values[1])

	def test_cr_note_against_invoice(self):
		ple = self.ple
		transaction_date = nowdate()
		amount = 100

		# reconcile against return invoice
		si3 = self.create_sales_invoice(qty=1, rate=amount, posting_date=transaction_date)
		cr_note1 = self.create_sales_invoice(
			qty=-1, rate=amount, posting_date=transaction_date, do_not_save=True, do_not_submit=True
		)
		cr_note1.is_return = 1
		cr_note1.return_against = si3.name
		cr_note1 = cr_note1.save().submit()

		pl_entries_si3 = (
			qb.from_(ple)
			.select(
				ple.voucher_type,
				ple.voucher_no,
				ple.against_voucher_type,
				ple.against_voucher_no,
				ple.amount,
				ple.delinked,
			)
			.where((ple.against_voucher_type == si3.doctype) & (ple.against_voucher_no == si3.name))
			.orderby(ple.creation)
			.run(as_dict=True)
		)

		pl_entries_cr_note1 = (
			qb.from_(ple)
			.select(
				ple.voucher_type,
				ple.voucher_no,
				ple.against_voucher_type,
				ple.against_voucher_no,
				ple.amount,
				ple.delinked,
			)
			.where((ple.against_voucher_type == cr_note1.doctype) & (ple.against_voucher_no == cr_note1.name))
			.orderby(ple.creation)
			.run(as_dict=True)
		)

		expected_values_for_si3 = [
			{
				"voucher_type": si3.doctype,
				"voucher_no": si3.name,
				"against_voucher_type": si3.doctype,
				"against_voucher_no": si3.name,
				"amount": amount,
				"delinked": 0,
			}
		]
		# credit/debit notes post ledger entries against itself
		expected_values_for_cr_note1 = [
			{
				"voucher_type": cr_note1.doctype,
				"voucher_no": cr_note1.name,
				"against_voucher_type": cr_note1.doctype,
				"against_voucher_no": cr_note1.name,
				"amount": -amount,
				"delinked": 0,
			},
		]
		self.assertEqual(pl_entries_si3, expected_values_for_si3)
		self.assertEqual(pl_entries_cr_note1, expected_values_for_cr_note1)

	def test_je_against_inv_and_note(self):
		ple = self.ple
		transaction_date = nowdate()
		amount = 100

		# reconcile against return invoice using JE
		si4 = self.create_sales_invoice(qty=1, rate=amount, posting_date=transaction_date)
		cr_note2 = self.create_sales_invoice(
			qty=-1, rate=amount, posting_date=transaction_date, do_not_save=True, do_not_submit=True
		)
		cr_note2.is_return = 1
		cr_note2 = cr_note2.save().submit()
		je1 = self.create_journal_entry(self.debit_to, self.debit_to, amount, posting_date=transaction_date)
		je1.get("accounts")[0].party_type = je1.get("accounts")[1].party_type = "Customer"
		je1.get("accounts")[0].party = je1.get("accounts")[1].party = self.customer
		je1.get("accounts")[0].reference_type = cr_note2.doctype
		je1.get("accounts")[0].reference_name = cr_note2.name
		je1.get("accounts")[1].reference_type = si4.doctype
		je1.get("accounts")[1].reference_name = si4.name
		je1 = je1.save().submit()

		pl_entries_for_invoice = (
			qb.from_(ple)
			.select(
				ple.voucher_type,
				ple.voucher_no,
				ple.against_voucher_type,
				ple.against_voucher_no,
				ple.amount,
				ple.delinked,
			)
			.where((ple.against_voucher_type == si4.doctype) & (ple.against_voucher_no == si4.name))
			.orderby(ple.creation)
			.run(as_dict=True)
		)

		expected_values = [
			{
				"voucher_type": si4.doctype,
				"voucher_no": si4.name,
				"against_voucher_type": si4.doctype,
				"against_voucher_no": si4.name,
				"amount": amount,
				"delinked": 0,
			},
			{
				"voucher_type": je1.doctype,
				"voucher_no": je1.name,
				"against_voucher_type": si4.doctype,
				"against_voucher_no": si4.name,
				"amount": -amount,
				"delinked": 0,
			},
		]
		self.assertEqual(pl_entries_for_invoice[0], expected_values[0])
		self.assertEqual(pl_entries_for_invoice[1], expected_values[1])

		pl_entries_for_crnote = (
			qb.from_(ple)
			.select(
				ple.voucher_type,
				ple.voucher_no,
				ple.against_voucher_type,
				ple.against_voucher_no,
				ple.amount,
				ple.delinked,
			)
			.where((ple.against_voucher_type == cr_note2.doctype) & (ple.against_voucher_no == cr_note2.name))
			.orderby(ple.creation)
			.run(as_dict=True)
		)

		expected_values = [
			{
				"voucher_type": cr_note2.doctype,
				"voucher_no": cr_note2.name,
				"against_voucher_type": cr_note2.doctype,
				"against_voucher_no": cr_note2.name,
				"amount": -amount,
				"delinked": 0,
			},
			{
				"voucher_type": je1.doctype,
				"voucher_no": je1.name,
				"against_voucher_type": cr_note2.doctype,
				"against_voucher_no": cr_note2.name,
				"amount": amount,
				"delinked": 0,
			},
		]
		self.assertEqual(pl_entries_for_crnote[0], expected_values[0])
		self.assertEqual(pl_entries_for_crnote[1], expected_values[1])

	@ERPNextTestSuite.change_settings(
		"Accounts Settings",
		{"unlink_payment_on_cancellation_of_invoice": 1, "delete_linked_ledger_entries": 1},
	)
	def test_multi_payment_unlink_on_invoice_cancellation(self):
		transaction_date = nowdate()
		amount = 100
		si = self.create_sales_invoice(qty=1, rate=amount, posting_date=transaction_date)

		for amt in [40, 40, 20]:
			# payment 1
			pe = get_payment_entry(si.doctype, si.name)
			pe.paid_amount = amt
			pe.get("references")[0].allocated_amount = amt
			pe = pe.save().submit()

		si.reload()
		si.cancel()

		entries = frappe.db.get_list(
			"Payment Ledger Entry",
			filters={"against_voucher_type": si.doctype, "against_voucher_no": si.name, "delinked": 0},
		)
		self.assertEqual(entries, [])

		# with references removed, deletion should be possible
		si.delete()
		self.assertRaises(frappe.DoesNotExistError, frappe.get_doc, si.doctype, si.name)

	@ERPNextTestSuite.change_settings(
		"Accounts Settings",
		{"unlink_payment_on_cancellation_of_invoice": 1, "delete_linked_ledger_entries": 1},
	)
	def test_multi_je_unlink_on_invoice_cancellation(self):
		transaction_date = nowdate()
		amount = 100
		si = self.create_sales_invoice(qty=1, rate=amount, posting_date=transaction_date)

		# multiple JE's against invoice
		for amt in [40, 40, 20]:
			je1 = self.create_journal_entry(
				self.income_account, self.debit_to, amt, posting_date=transaction_date
			)
			je1.get("accounts")[1].party_type = "Customer"
			je1.get("accounts")[1].party = self.customer
			je1.get("accounts")[1].reference_type = si.doctype
			je1.get("accounts")[1].reference_name = si.name
			je1 = je1.save().submit()

		si.reload()
		si.cancel()

		entries = frappe.db.get_list(
			"Payment Ledger Entry",
			filters={"against_voucher_type": si.doctype, "against_voucher_no": si.name, "delinked": 0},
		)
		self.assertEqual(entries, [])

		# with references removed, deletion should be possible
		si.delete()
		self.assertRaises(frappe.DoesNotExistError, frappe.get_doc, si.doctype, si.name)

	@ERPNextTestSuite.change_settings(
		"Accounts Settings",
		{
			"unlink_payment_on_cancellation_of_invoice": 1,
			"delete_linked_ledger_entries": 1,
			"unlink_advance_payment_on_cancelation_of_order": 1,
		},
	)
	def test_advance_payment_unlink_on_order_cancellation(self):
		transaction_date = nowdate()
		amount = 100
		so = self.create_sales_order(qty=1, rate=amount, posting_date=transaction_date).save().submit()

		get_payment_entry(so.doctype, so.name).save().submit()

		so.reload()
		so.cancel()

		entries = frappe.db.get_list(
			"Payment Ledger Entry",
			filters={"against_voucher_type": so.doctype, "against_voucher_no": so.name, "delinked": 0},
		)
		self.assertEqual(entries, [])

		# with references removed, deletion should be possible
		so.delete()
		self.assertRaises(frappe.DoesNotExistError, frappe.get_doc, so.doctype, so.name)

	@ERPNextTestSuite.change_settings(
		"Accounts Settings",
		{"enable_immutable_ledger": 1},
	)
	def test_reverse_entries_on_cancel_for_immutable_ledger(self):
		invoice_posting_date = add_days(nowdate(), -5)
		gle = qb.DocType("GL Entry")
		ple = qb.DocType("Payment Ledger Entry")

		si = self.create_sales_invoice(qty=1, rate=100, posting_date=invoice_posting_date)

		gles_before = (
			qb.from_(gle)
			.select(
				Count(gle.name),
			)
			.where((gle.voucher_type == si.doctype) & (gle.voucher_no == si.name) & (gle.is_cancelled == 0))
			.run()[0][0]
		)
		ples_before = (
			qb.from_(ple)
			.select(
				Count(ple.name),
			)
			.where((ple.voucher_type == si.doctype) & (ple.voucher_no == si.name) & (ple.delinked.eq(0)))
			.run()[0][0]
		)

		si.cancel()

		gles_after = (
			qb.from_(gle)
			.select(Count(gle.account))
			.where((gle.voucher_type == si.doctype) & (gle.voucher_no == si.name) & (gle.is_cancelled == 0))
			.run()[0][0]
		)
		self.assertEqual(gles_after, gles_before * 2)

		ples_after = (
			qb.from_(ple)
			.select(
				Count(ple.name),
			)
			.where((ple.voucher_type == si.doctype) & (ple.voucher_no == si.name) & (ple.delinked.eq(0)))
			.run()[0][0]
		)
		self.assertEqual(ples_after, ples_before * 2)

		# assert debit/credit are reversed
		gl_entries = (
			qb.from_(gle)
			.select(gle.account, Sum(gle.debit).as_("total_debit"), Sum(gle.credit).as_("total_credit"))
			.where((gle.voucher_type == si.doctype) & (gle.voucher_no == si.name) & (gle.is_cancelled == 0))
			.groupby(gle.account)
			.run(as_dict=True)
		)
		for gl in gl_entries:
			with self.subTest(gl=gl):
				self.assertEqual(gl.total_debit, gl.total_credit)

		# assert amounts are reversed
		pl_entries = (
			qb.from_(ple)
			.select(ple.account, Sum(ple.amount).as_("total_amount"))
			.where((ple.voucher_type == si.doctype) & (ple.voucher_no == si.name) & (ple.delinked == 0))
			.groupby(ple.account)
			.run(as_dict=True)
		)
		for pl in pl_entries:
			with self.subTest(pl=pl):
				self.assertEqual(pl.total_amount, 0)

		self.assertFalse(
			frappe.db.exists(
				"Payment Ledger Entry",
				{"voucher_type": si.doctype, "voucher_no": si.name, "delinked": 1},
			)
		)
