mirror of
https://github.com/frappe/erpnext.git
synced 2026-08-14 23:18:40 +00:00
Merge pull request #56083 from nabinhait/refactor-si-pos-service
refactor(sales_invoice): tidy POSService (set_pos_fields + mode-of-payment queries)
This commit is contained in:
Nabin Hait
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user