fix(general-ledger): add CPA support via Party Link cross-party GL filtering

erpnext/accounts/report/general_ledger/general_ledger.py

@@ -95,6 +95,9 @@ def validate_filters(filters, account_details):
 def validate_party(filters):
 	party_type, party = filters.get("party_type"), filters.get("party")
 
+	if party and not party_type:
+		frappe.throw(_("Please select a Party Type before selecting a Party."))
+
 	if party and party_type:
 		for d in party:
 			if not frappe.db.exists(party_type, d):
@@ -277,11 +280,10 @@ def get_conditions(filters):
 	if filters.get("categorize_by") == "Categorize by Party" and not filters.get("party_type"):
 		conditions.append("party_type in ('Customer', 'Supplier')")
 
-	if filters.get("party_type"):
-		conditions.append("party_type=%(party_type)s")
-
 	if filters.get("party"):
-		conditions.append("party in %(party)s")
+		conditions.append(build_common_party_condition(filters))
+	elif filters.get("party_type"):
+		conditions.append("party_type=%(party_type)s")
 
 	if not (
 		filters.get("account")
@@ -345,6 +347,44 @@ def get_conditions(filters):
 	return "and {}".format(" and ".join(conditions)) if conditions else ""
 
 
+def get_linked_parties(party_list: list, party_type: str) -> dict:
+	parties_by_type = {party_type: list(party_list)}
+
+	party_link_dt = frappe.qb.DocType("Party Link")
+	links = (
+		frappe.qb.from_(party_link_dt)
+		.select(
+			party_link_dt.primary_role,
+			party_link_dt.primary_party,
+			party_link_dt.secondary_role,
+			party_link_dt.secondary_party,
+		)
+		.where(
+			(party_link_dt.secondary_party.isin(party_list)) | (party_link_dt.primary_party.isin(party_list))
+		)
+		.run(as_dict=True)
+	)
+
+	for link in links:
+		if link.secondary_party in party_list and link.secondary_role == party_type:
+			parties_by_type.setdefault(link.primary_role, []).append(link.primary_party)
+		elif link.primary_party in party_list and link.primary_role == party_type:
+			parties_by_type.setdefault(link.secondary_role, []).append(link.secondary_party)
+
+	return parties_by_type
+
+
+def build_common_party_condition(filters):
+	parties_by_type = get_linked_parties(list(filters.get("party")), filters.get("party_type"))
+
+	parts = []
+	for ptype, p_list in parties_by_type.items():
+		escaped_parties = ", ".join(frappe.db.escape(p) for p in p_list)
+		parts.append(f"(party_type = {frappe.db.escape(ptype)} AND party IN ({escaped_parties}))")
+
+	return "(" + " OR ".join(parts) + ")"
+
+
 def get_party_name_map():
 	party_map = {}
 

erpnext/accounts/report/general_ledger/test_general_ledger.py

@@ -5,6 +5,7 @@
 from frappe import qb
 from frappe.utils import flt, today
 
+from erpnext.accounts.doctype.party_link.party_link import create_party_link
 from erpnext.accounts.doctype.sales_invoice.test_sales_invoice import create_sales_invoice
 from erpnext.accounts.report.general_ledger.general_ledger import execute
 from erpnext.controllers.sales_and_purchase_return import make_return_doc
@@ -28,6 +29,176 @@ def clear_old_entries(self):
 		for doctype in doctype_list:
 			qb.from_(qb.DocType(doctype)).delete().where(qb.DocType(doctype).company == self.company).run()
 
+	def test_cpa_includes_linked_party_entries(self):
+		"""
+		CPA support: filtering GL by a Customer should also return GL entries posted under the linked Supplier (and vice versa) via Party Link.
+		"""
+		company = self.company
+		customer = "_Test Customer"
+		supplier = "_Test Supplier"
+
+		# clean up any existing Party Link involving either party on either side
+		if pl_name := frappe.get_all(
+			"Party Link",
+			or_filters=[
+				["primary_party", "in", [customer, supplier]],
+				["secondary_party", "in", [customer, supplier]],
+			],
+			pluck="name",
+		):
+			frappe.delete_doc("Party Link", pl_name[0], ignore_permissions=True)
+
+		party_link = create_party_link(
+			primary_role="Customer",
+			primary_party=customer,
+			secondary_party=supplier,
+		)
+
+		try:
+			# GL entry via Sales Invoice (Customer side)
+			si = create_sales_invoice(customer=customer, company=company, do_not_submit=False)
+
+			# GL entry via Purchase Invoice (Supplier side)
+			from erpnext.accounts.doctype.purchase_invoice.test_purchase_invoice import make_purchase_invoice
+
+			pi = make_purchase_invoice(supplier=supplier, company=company, do_not_submit=False)
+
+			# Filter by Customer only — should include both Customer and Supplier entries
+			_, data = execute(
+				frappe._dict(
+					{
+						"company": company,
+						"from_date": si.posting_date,
+						"to_date": si.posting_date,
+						"party_type": "Customer",
+						"party": [customer],
+					}
+				)
+			)
+
+			party_types_in_result = {row.party_type for row in data if row.get("party_type")}
+			self.assertIn("Customer", party_types_in_result)
+			self.assertIn("Supplier", party_types_in_result)
+
+			# Filter by Supplier only — should include both sides too
+			_, data = execute(
+				frappe._dict(
+					{
+						"company": company,
+						"from_date": pi.posting_date,
+						"to_date": pi.posting_date,
+						"party_type": "Supplier",
+						"party": [supplier],
+					}
+				)
+			)
+
+			party_types_in_result = {row.party_type for row in data if row.get("party_type")}
+			self.assertIn("Customer", party_types_in_result)
+			self.assertIn("Supplier", party_types_in_result)
+		finally:
+			frappe.delete_doc("Party Link", party_link.name, ignore_permissions=True)
+
+	def test_party_filter_without_party_link(self):
+		"""
+		When a party has no Party Link, GL filtering should only return entries for that party — no cross-party leakage.
+		"""
+		company = self.company
+		customer = "_Test Customer"
+		other_customer = "_Test Customer 1"
+
+		# Ensure no Party Link exists for these parties (either side)
+		if pl_name := frappe.get_all(
+			"Party Link",
+			or_filters=[
+				["primary_party", "in", [customer, other_customer]],
+				["secondary_party", "in", [customer, other_customer]],
+			],
+			pluck="name",
+		):
+			frappe.delete_doc("Party Link", pl_name[0], ignore_permissions=True)
+
+		# GL entry for customer
+		si = create_sales_invoice(customer=customer, company=company, do_not_submit=False)
+		# GL entry for a different customer
+		create_sales_invoice(customer=other_customer, company=company, do_not_submit=False)
+
+		_, data = execute(
+			frappe._dict(
+				{
+					"company": company,
+					"from_date": si.posting_date,
+					"to_date": si.posting_date,
+					"party_type": "Customer",
+					"party": [customer],
+				}
+			)
+		)
+
+		parties_in_result = {row.party for row in data if row.get("party")}
+		self.assertIn(customer, parties_in_result)
+		self.assertNotIn(other_customer, parties_in_result)
+
+	def test_cpa_mixed_party_list_partial_links(self):
+		"""
+		Linked counterparts are included only for parties that have a Party Link; unlinked parties return only their own GL entries.
+		"""
+		company = self.company
+		linked_customer = "_Test Customer"
+		unlinked_customer = "_Test Customer 1"
+		supplier = "_Test Supplier"
+
+		# clean up any existing Party Link touching these parties
+		if pl_name := frappe.get_all(
+			"Party Link",
+			or_filters=[
+				["primary_party", "in", [linked_customer, unlinked_customer, supplier]],
+				["secondary_party", "in", [linked_customer, unlinked_customer, supplier]],
+			],
+			pluck="name",
+		):
+			frappe.delete_doc("Party Link", pl_name[0], ignore_permissions=True)
+
+		party_link = create_party_link(
+			primary_role="Customer",
+			primary_party=linked_customer,
+			secondary_party=supplier,
+		)
+
+		try:
+			from erpnext.accounts.doctype.purchase_invoice.test_purchase_invoice import make_purchase_invoice
+
+			si_linked = create_sales_invoice(customer=linked_customer, company=company, do_not_submit=False)
+			_si_unlinked = create_sales_invoice(
+				customer=unlinked_customer, company=company, do_not_submit=False
+			)
+			make_purchase_invoice(supplier=supplier, company=company, do_not_submit=False)
+
+			_, data = execute(
+				frappe._dict(
+					{
+						"company": company,
+						"from_date": si_linked.posting_date,
+						"to_date": si_linked.posting_date,
+						"party_type": "Customer",
+						"party": [linked_customer, unlinked_customer],
+					}
+				)
+			)
+
+			parties_in_result = {row.party for row in data if row.get("party")}
+			party_types_in_result = {row.party_type for row in data if row.get("party_type")}
+
+			# linked_customer's own entries are present
+			self.assertIn(linked_customer, parties_in_result)
+			# unlinked_customer's own entries are present
+			self.assertIn(unlinked_customer, parties_in_result)
+			# supplier linked to linked_customer is pulled in
+			self.assertIn(supplier, parties_in_result)
+			self.assertIn("Supplier", party_types_in_result)
+		finally:
+			frappe.delete_doc("Party Link", party_link.name, ignore_permissions=True)
+
 	def test_foreign_account_balance_after_exchange_rate_revaluation(self):
 		"""
 		Checks the correctness of balance after exchange rate revaluation