Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 63 additions & 0 deletions pos_next/api/invoices.py
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,60 @@ def validate_manual_rate_edit(item, pos_profile=None, pos_settings_cache=None):
return {"valid": True}


def _clean_payment_row(payment):
"""Return a child-row-safe payment dict without parent metadata."""
if hasattr(payment, "as_dict"):
payment_data = payment.as_dict()
elif isinstance(payment, dict):
payment_data = dict(payment)
else:
payment_data = dict(getattr(payment, "__dict__", {}))

fields_to_strip = {
"doctype",
"parent",
"parentfield",
"parenttype",
"owner",
"creation",
"modified",
"modified_by",
"__last_sync_on",
"__unsaved",
}
return {key: value for key, value in payment_data.items() if key not in fields_to_strip}


def _snapshot_invoice_payments(invoice_doc):
"""Capture the payment rows before ERPNext POS defaults overwrite them."""
return [_clean_payment_row(payment) for payment in (invoice_doc.get("payments") or [])]


def _restore_invoice_payments(invoice_doc, payments):
"""Restore exactly the payment rows provided by the POS client."""
invoice_doc.set("payments", [])
for payment in payments:
invoice_doc.append("payments", payment)


def _sync_invoice_payment_amounts(invoice_doc):
"""Recompute payment totals after restoring cashier-entered payment rows."""
conversion_rate = flt(invoice_doc.get("conversion_rate")) or 1
paid_amount = 0
base_paid_amount = 0

for payment in invoice_doc.get("payments") or []:
payment.amount = flt(payment.get("amount") or 0)
payment.base_amount = flt(payment.get("base_amount") or 0) or flt(
payment.amount * conversion_rate
)
paid_amount += payment.amount
base_paid_amount += payment.base_amount

invoice_doc.paid_amount = flt(paid_amount)
invoice_doc.base_paid_amount = flt(base_paid_amount)


def log_manual_rate_edit(item, invoice_name, user=None):
"""
Create an audit log entry for manual rate edits.
Expand Down Expand Up @@ -848,6 +902,9 @@ def update_invoice(data):
invoice_doc.update_stock = 1
if pos_profile_doc and pos_profile_doc.warehouse:
invoice_doc.set_warehouse = pos_profile_doc.warehouse
incoming_payments = _snapshot_invoice_payments(invoice_doc)
else:
incoming_payments = []

# ========================================================================
# ROUNDING CONFIGURATION
Expand All @@ -866,6 +923,9 @@ def update_invoice(data):
# Populate missing fields (company, currency, accounts, etc.)
invoice_doc.set_missing_values()

if doctype == "Sales Invoice":
_restore_invoice_payments(invoice_doc, incoming_payments)

# Calculate totals and apply discounts (with rounding disabled)
invoice_doc.calculate_taxes_and_totals()
if invoice_doc.grand_total is None:
Expand All @@ -876,6 +936,9 @@ def update_invoice(data):
# Set accounts for payment methods before saving
_set_payment_accounts(invoice_doc.payments, invoice_doc.company)

if doctype == "Sales Invoice":
_sync_invoice_payment_amounts(invoice_doc)

# For return invoices, ensure payments are negative
if invoice_doc.get("is_return"):
# Return handling is primarily for Sales Invoice
Expand Down
104 changes: 104 additions & 0 deletions pos_next/api/test_invoices.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
# Copyright (c) 2026, MT
# See license.txt

import unittest

from pos_next.api.invoices import (
_restore_invoice_payments,
_snapshot_invoice_payments,
_sync_invoice_payment_amounts,
)


class _FakePayment(dict):
def __getattr__(self, key):
return self.get(key)

def __setattr__(self, key, value):
self[key] = value

def as_dict(self):
return dict(self)


class _FakeInvoice:
def __init__(self, payments=None, conversion_rate=1):
self.payments = [_FakePayment(payment) for payment in (payments or [])]
self.conversion_rate = conversion_rate
self.paid_amount = 0
self.base_paid_amount = 0

def get(self, key, default=None):
return getattr(self, key, default)

def set(self, key, value):
if key == "payments":
self.payments = [_FakePayment(payment) for payment in value]
else:
setattr(self, key, value)

def append(self, key, value):
if key != "payments":
raise AssertionError("Only payments are supported in this fake doc")
self.payments.append(_FakePayment(value))


class TestInvoicePayments(unittest.TestCase):
def test_payment_snapshot_and_restore_preserve_amounts(self):
invoice = _FakeInvoice(
payments=[
{
"name": "row-1",
"mode_of_payment": "Cash",
"amount": 200,
"base_amount": 200,
"account": "1110 - Cash - S",
"parent": "ACC-SINV-0001",
"doctype": "Sales Invoice Payment",
}
]
)

snapshot = _snapshot_invoice_payments(invoice)

invoice.set(
"payments",
[
{
"mode_of_payment": "Cash",
"amount": 0,
"base_amount": 0,
},
{
"mode_of_payment": "mBok",
"amount": 0,
"base_amount": 0,
},
],
)

_restore_invoice_payments(invoice, snapshot)

assert len(invoice.payments) == 1
assert invoice.payments[0].mode_of_payment == "Cash"
assert invoice.payments[0].amount == 200
assert invoice.payments[0].base_amount == 200
assert invoice.payments[0].account == "1110 - Cash - S"
assert "parent" not in invoice.payments[0]
assert "doctype" not in invoice.payments[0]

def test_sync_invoice_payment_amounts_uses_conversion_rate(self):
invoice = _FakeInvoice(
payments=[
{"mode_of_payment": "Cash", "amount": 100},
{"mode_of_payment": "Card", "amount": 50, "base_amount": 110},
],
conversion_rate=2,
)

_sync_invoice_payment_amounts(invoice)

assert invoice.payments[0].base_amount == 200
assert invoice.payments[1].base_amount == 110
assert invoice.paid_amount == 150
assert invoice.base_paid_amount == 310
Loading