From 7f47c218ceff59b68b966484983254191d4b0ac1 Mon Sep 17 00:00:00 2001 From: Nabin Hait Date: Thu, 18 Jun 2026 13:43:12 +0530 Subject: [PATCH 1/2] test(sales_invoice): characterize POSService default and mode-of-payment logic Pin the behaviour of POSService.set_pos_fields (POS-profile default resolution and the for_validate guard) and the mode-of-payment query helpers before refactoring them. --- .../sales_invoice/test_sales_invoice.py | 101 ++++++++++++++++++ 1 file changed, 101 insertions(+) diff --git a/erpnext/accounts/doctype/sales_invoice/test_sales_invoice.py b/erpnext/accounts/doctype/sales_invoice/test_sales_invoice.py index e168a9e9df1..aa1a969b59b 100644 --- a/erpnext/accounts/doctype/sales_invoice/test_sales_invoice.py +++ b/erpnext/accounts/doctype/sales_invoice/test_sales_invoice.py @@ -20,6 +20,12 @@ from erpnext.accounts.doctype.purchase_invoice.test_purchase_invoice import ( unlink_payment_on_cancel_of_invoice, ) from erpnext.accounts.doctype.sales_invoice.mapper import make_inter_company_transaction +from erpnext.accounts.doctype.sales_invoice.services.pos import ( + POSService, + get_all_mode_of_payments, + get_mode_of_payment_info, + get_mode_of_payments_info, +) from erpnext.accounts.utils import PaymentEntryUnlinkError from erpnext.assets.doctype.asset.depreciation import post_depreciation_entries from erpnext.assets.doctype.asset.test_asset import create_asset @@ -1346,6 +1352,101 @@ class TestSalesInvoice(ERPNextTestSuite): self.assertEqual(pos.grand_total, 100.0) self.assertEqual(pos.write_off_amount, 0) + def test_set_pos_fields_populates_invoice_from_profile(self): + terms = frappe.db.exists("Terms and Conditions", "_Test POS Terms") + if not terms: + terms = ( + frappe.get_doc( + { + "doctype": "Terms and Conditions", + "title": "_Test POS Terms", + "terms": "POS terms and conditions", + "selling": 1, + } + ) + .insert() + .name + ) + + profile = make_pos_profile() + profile.customer = "_Test Customer" + profile.tax_category = "_Test Tax Category 1" + profile.account_for_change_amount = "Cash - _TC" + profile.ignore_pricing_rule = 1 + profile.update_stock = 1 + profile.apply_discount_on = "Grand Total" + profile.tc_name = terms + profile.taxes_and_charges = "_Test Sales Taxes and Charges Template - _TC" + profile.save() + + si = create_sales_invoice(do_not_save=True) + si.is_pos = 1 + si.pos_profile = profile.name + si.customer = None + si.taxes = [] + + POSService(si).set_pos_fields(for_validate=False) + + self.assertEqual(si.customer, "_Test Customer") + self.assertEqual(si.tax_category, "_Test Tax Category 1") + self.assertEqual(si.ignore_pricing_rule, 1) + self.assertEqual(si.account_for_change_amount, "Cash - _TC") + self.assertEqual(si.taxes_and_charges, "_Test Sales Taxes and Charges Template - _TC") + self.assertEqual(si.apply_discount_on, "Grand Total") + self.assertEqual(si.update_stock, 1) + self.assertEqual(si.terms, "POS terms and conditions") + self.assertTrue(si.get("payments")) + self.assertTrue(si.get("taxes")) + + def test_set_pos_fields_for_validate_preserves_existing_values(self): + profile = make_pos_profile() + profile.tax_category = "_Test Tax Category 1" + profile.save() + + si = create_sales_invoice(do_not_save=True) + si.is_pos = 1 + si.pos_profile = profile.name + si.apply_discount_on = "Net Total" + existing_customer = si.customer + + POSService(si).set_pos_fields(for_validate=True) + + # for_validate must not overwrite a field the user already set + self.assertEqual(si.apply_discount_on, "Net Total") + # for_validate skips mode-of-payment fetch and profile-driven customer/tax_category + self.assertFalse(si.get("payments")) + self.assertEqual(si.customer, existing_customer) + self.assertFalse(si.tax_category) + + def test_set_pos_fields_uses_profile_price_list_without_customer(self): + profile = make_pos_profile(selling_price_list="_Test Price List") + profile.customer = None + profile.save() + + si = create_sales_invoice(do_not_save=True) + si.is_pos = 1 + si.pos_profile = profile.name + si.customer = None + + POSService(si).set_pos_fields(for_validate=False) + + self.assertEqual(si.selling_price_list, "_Test Price List") + + def test_pos_service_mode_of_payment_queries(self): + make_pos_profile() # ensures a Cash mode-of-payment account for _Test Company + si = create_sales_invoice(do_not_save=True) + + single = get_mode_of_payment_info("Cash", "_Test Company") + self.assertTrue(single) + self.assertEqual(single[0].parent, "Cash") + + all_modes = get_all_mode_of_payments(si) + self.assertTrue(any(row.parent == "Cash" for row in all_modes)) + + grouped = get_mode_of_payments_info(["Cash"], "_Test Company") + self.assertIn("Cash", grouped) + self.assertEqual(grouped["Cash"].mop, "Cash") + def test_auto_write_off_amount(self): make_pos_profile( company="_Test Company with perpetual inventory", From 9f02c4759270c59eec935b161d521aac0040ea70 Mon Sep 17 00:00:00 2001 From: Nabin Hait Date: Thu, 18 Jun 2026 13:43:25 +0530 Subject: [PATCH 2/2] refactor(sales_invoice): decompose POSService.set_pos_fields and dedupe MOP queries set_pos_fields was a 97-line method (cyclomatic complexity E/36). Split it into a small orchestrator plus focused helpers, each preserving the exact for_validate semantics (A/5 after). Collapse the three near-identical mode-of-payment query builders onto a shared _enabled_mode_of_payment_query, and add type hints and docstrings. Public signatures and return shapes are unchanged. --- .../doctype/sales_invoice/services/pos.py | 261 ++++++++++-------- 1 file changed, 140 insertions(+), 121 deletions(-) diff --git a/erpnext/accounts/doctype/sales_invoice/services/pos.py b/erpnext/accounts/doctype/sales_invoice/services/pos.py index 7fa28220fb3..0e596042d08 100644 --- a/erpnext/accounts/doctype/sales_invoice/services/pos.py +++ b/erpnext/accounts/doctype/sales_invoice/services/pos.py @@ -13,106 +13,140 @@ class PartialPaymentValidationError(frappe.ValidationError): class POSService: - def __init__(self, doc): + def __init__(self, doc) -> None: self.doc = doc - def set_pos_fields(self, for_validate: bool = False) -> frappe.Document | None: - """Populate POS-profile fields on the invoice; return the profile or None.""" + def set_pos_fields(self, for_validate: bool = False) -> frappe.Document | dict | None: + """Populate POS-profile fields on the invoice; return the profile, {} or None.""" doc = self.doc if cint(doc.is_pos) != 1: return None + self._set_default_change_amount_account() + + if not self._ensure_pos_profile(): + return None + + pos = frappe.get_doc("POS Profile", doc.pos_profile) if doc.pos_profile else {} + if pos: + self._apply_pos_profile(pos, for_validate) + + return pos + + def _set_default_change_amount_account(self) -> None: + doc = self.doc if not doc.account_for_change_amount: doc.account_for_change_amount = frappe.get_cached_value( "Company", doc.company, "default_cash_account" ) - from erpnext.stock.get_item_details import ( - ItemDetailsCtx, - get_pos_profile, - get_pos_profile_item_details_, - ) + def _ensure_pos_profile(self) -> bool: + """Auto-pick a POS Profile for the company; return False if none could be found.""" + doc = self.doc + if doc.pos_profile or doc.flags.ignore_pos_profile: + return True - if not doc.pos_profile and not doc.flags.ignore_pos_profile: - pos_profile = get_pos_profile(doc.company) or {} - if not pos_profile: - return None - doc.pos_profile = pos_profile.get("name") + from erpnext.stock.get_item_details import get_pos_profile - pos = {} - if doc.pos_profile: - pos = frappe.get_doc("POS Profile", doc.pos_profile) + pos_profile = get_pos_profile(doc.company) or {} + if not pos_profile: + return False - if pos: - if not for_validate: - update_multi_mode_option(doc, pos) - doc.tax_category = pos.get("tax_category") + doc.pos_profile = pos_profile.get("name") + return True - if not for_validate and not doc.customer: - doc.customer = pos.customer + def _apply_pos_profile(self, pos, for_validate: bool) -> None: + doc = self.doc + if not for_validate: + self._apply_editable_pos_defaults(pos) - if not for_validate: - doc.ignore_pricing_rule = pos.ignore_pricing_rule + if pos.get("account_for_change_amount"): + doc.account_for_change_amount = pos.get("account_for_change_amount") - if pos.get("account_for_change_amount"): - doc.account_for_change_amount = pos.get("account_for_change_amount") + self._copy_pos_profile_fields(pos, for_validate) - for fieldname in ( - "currency", - "letter_head", - "tc_name", - "company", - "select_print_heading", - "write_off_account", - "taxes_and_charges", - "write_off_cost_center", - "apply_discount_on", - "cost_center", - ): - if (not for_validate) or (for_validate and not doc.get(fieldname)): - doc.set(fieldname, pos.get(fieldname)) + if pos.get("company_address"): + doc.company_address = pos.get("company_address") - if pos.get("company_address"): - doc.company_address = pos.get("company_address") + self._set_selling_price_list(pos) - if doc.customer: - customer_price_list, customer_group = frappe.get_value( - "Customer", doc.customer, ["default_price_list", "customer_group"] - ) - customer_group_price_list = frappe.get_value( - "Customer Group", customer_group, "default_price_list" - ) - selling_price_list = ( - customer_price_list or customer_group_price_list or pos.get("selling_price_list") - ) - else: - selling_price_list = pos.get("selling_price_list") + if not for_validate: + self._set_update_stock_from_profile(pos) - if selling_price_list: - doc.set("selling_price_list", selling_price_list) + self._apply_pos_item_defaults(pos, for_validate) + self._set_terms_and_taxes(pos) - if not for_validate: - dn_flag = any(d.get("dn_detail") for d in doc.get("items")) - doc.update_stock = 0 if dn_flag else cint(pos.get("update_stock")) + def _apply_editable_pos_defaults(self, pos) -> None: + """Profile defaults the user may override; only applied outside validation.""" + doc = self.doc + update_multi_mode_option(doc, pos) + doc.tax_category = pos.get("tax_category") + if not doc.customer: + doc.customer = pos.customer + doc.ignore_pricing_rule = pos.ignore_pricing_rule - for item in doc.get("items"): - if item.get("item_code"): - profile_details = get_pos_profile_item_details_( - ItemDetailsCtx(item.as_dict()), pos, pos, update_data=True - ) - for fname, val in profile_details.items(): - if (not for_validate) or (for_validate and not item.get(fname)): - item.set(fname, val) + def _copy_pos_profile_fields(self, pos, for_validate: bool) -> None: + doc = self.doc + for fieldname in ( + "currency", + "letter_head", + "tc_name", + "company", + "select_print_heading", + "write_off_account", + "taxes_and_charges", + "write_off_cost_center", + "apply_discount_on", + "cost_center", + ): + if (not for_validate) or (for_validate and not doc.get(fieldname)): + doc.set(fieldname, pos.get(fieldname)) - if doc.tc_name and not doc.terms: - doc.terms = frappe.db.get_value("Terms and Conditions", doc.tc_name, "terms") + def _set_selling_price_list(self, pos) -> None: + doc = self.doc + if doc.customer: + customer_price_list, customer_group = frappe.get_value( + "Customer", doc.customer, ["default_price_list", "customer_group"] + ) + customer_group_price_list = frappe.get_value( + "Customer Group", customer_group, "default_price_list" + ) + selling_price_list = ( + customer_price_list or customer_group_price_list or pos.get("selling_price_list") + ) + else: + selling_price_list = pos.get("selling_price_list") - if doc.taxes_and_charges and not len(doc.get("taxes")): - from erpnext.accounts.services.taxes import TaxService + if selling_price_list: + doc.set("selling_price_list", selling_price_list) - TaxService(doc).set_taxes() + def _set_update_stock_from_profile(self, pos) -> None: + doc = self.doc + dn_flag = any(d.get("dn_detail") for d in doc.get("items")) + doc.update_stock = 0 if dn_flag else cint(pos.get("update_stock")) - return pos + def _apply_pos_item_defaults(self, pos, for_validate: bool) -> None: + from erpnext.stock.get_item_details import ItemDetailsCtx, get_pos_profile_item_details_ + + for item in self.doc.get("items"): + if not item.get("item_code"): + continue + profile_details = get_pos_profile_item_details_( + ItemDetailsCtx(item.as_dict()), pos, pos, update_data=True + ) + for fname, val in profile_details.items(): + if (not for_validate) or (for_validate and not item.get(fname)): + item.set(fname, val) + + def _set_terms_and_taxes(self, pos) -> None: + doc = self.doc + if doc.tc_name and not doc.terms: + doc.terms = frappe.db.get_value("Terms and Conditions", doc.tc_name, "terms") + + if doc.taxes_and_charges and not len(doc.get("taxes")): + from erpnext.accounts.services.taxes import TaxService + + TaxService(doc).set_taxes() def update_paid_amount(self) -> None: doc = self.doc @@ -144,6 +178,7 @@ class POSService: doc.paid_amount = 0 def validate_pos_return(self) -> None: + """Ensure POS return payments are not less than the (negative) invoice total.""" doc = self.doc if doc.is_consolidated: return @@ -160,6 +195,7 @@ class POSService: frappe.throw(_("At least one mode of payment is required for POS invoice.")) def validate_pos(self) -> None: + """On a POS return, paid amount plus write-off cannot exceed the grand total.""" doc = self.doc if doc.is_return: invoice_total = doc.rounded_total or doc.grand_total @@ -180,6 +216,7 @@ class POSService: self.validate_pos_opening_entry() def validate_full_payment(self) -> None: + """Block partial payment on a submitted POS invoice unless the profile allows it.""" doc = self.doc allow_partial_payment = frappe.db.get_value("POS Profile", doc.pos_profile, "allow_partial_payment") invoice_total = flt(doc.rounded_total) or flt(doc.grand_total) @@ -196,6 +233,7 @@ class POSService: ) def validate_pos_opening_entry(self) -> None: + """Require exactly one current, open POS Opening Entry for the profile.""" doc = self.doc opening_entries = frappe.get_all( "POS Opening Entry", @@ -337,62 +375,43 @@ def update_multi_mode_option(doc, pos_profile) -> None: def get_all_mode_of_payments(doc) -> list: - ModeOfPaymentAccount = frappe.qb.DocType("Mode of Payment Account") - ModeOfPayment = frappe.qb.DocType("Mode of Payment") - - query = ( - frappe.qb.from_(ModeOfPaymentAccount) - .join(ModeOfPayment) - .on(ModeOfPaymentAccount.parent == ModeOfPayment.name) - .select( - ModeOfPaymentAccount.default_account, ModeOfPaymentAccount.parent, ModeOfPayment.type.as_("type") - ) - .where(ModeOfPaymentAccount.company == doc.company) - .where(ModeOfPayment.enabled == 1) - ) - - return query.run(as_dict=1) + """All enabled modes of payment with their default accounts for the doc's company.""" + query, mopa, mop = _enabled_mode_of_payment_query(doc.company) + return query.select(mopa.default_account, mopa.parent, mop.type.as_("type")).run(as_dict=1) def get_mode_of_payments_info(mode_of_payments: list, company: str) -> dict: - ModeOfPaymentAccount = frappe.qb.DocType("Mode of Payment Account") - ModeOfPayment = frappe.qb.DocType("Mode of Payment") - - query = ( - frappe.qb.from_(ModeOfPaymentAccount) - .join(ModeOfPayment) - .on(ModeOfPaymentAccount.parent == ModeOfPayment.name) - .select( - ModeOfPaymentAccount.default_account, - ModeOfPaymentAccount.parent.as_("mop"), - ModeOfPayment.type.as_("type"), - ) - .where(ModeOfPaymentAccount.company == company) - .where(ModeOfPayment.enabled == 1) - .where(ModeOfPayment.name.isin(mode_of_payments)) + """Map each of the named modes of payment to its account info for the company.""" + query, mopa, mop = _enabled_mode_of_payment_query(company) + data = ( + query.select(mopa.default_account, mopa.parent.as_("mop"), mop.type.as_("type")) + .where(mop.name.isin(mode_of_payments)) # group by all selected columns so postgres accepts it (one row per mode of payment) - .groupby(ModeOfPaymentAccount.default_account, ModeOfPaymentAccount.parent, ModeOfPayment.type) + .groupby(mopa.default_account, mopa.parent, mop.type) + .run(as_dict=1) ) - - data = query.run(as_dict=1) - return {row.get("mop"): row for row in data} def get_mode_of_payment_info(mode_of_payment: str, company: str) -> list: - ModeOfPaymentAccount = frappe.qb.DocType("Mode of Payment Account") - ModeOfPayment = frappe.qb.DocType("Mode of Payment") - - query = ( - frappe.qb.from_(ModeOfPayment) - .join(ModeOfPaymentAccount) - .on(ModeOfPaymentAccount.parent == ModeOfPayment.name) - .select( - ModeOfPaymentAccount.default_account, ModeOfPaymentAccount.parent, ModeOfPayment.type.as_("type") - ) - .where(ModeOfPaymentAccount.company == company) - .where(ModeOfPayment.enabled == 1) - .where(ModeOfPayment.name == mode_of_payment) + """Account info for a single mode of payment in the company.""" + query, mopa, mop = _enabled_mode_of_payment_query(company) + return ( + query.select(mopa.default_account, mopa.parent, mop.type.as_("type")) + .where(mop.name == mode_of_payment) + .run(as_dict=1) ) - return query.run(as_dict=1) + +def _enabled_mode_of_payment_query(company: str): + """Base query joining enabled modes of payment to their accounts for a company.""" + mopa = frappe.qb.DocType("Mode of Payment Account") + mop = frappe.qb.DocType("Mode of Payment") + query = ( + frappe.qb.from_(mopa) + .join(mop) + .on(mopa.parent == mop.name) + .where(mopa.company == company) + .where(mop.enabled == 1) + ) + return query, mopa, mop