Skip to content

[MIG] accoutn_ebics_payment_return#280

Draft
ecino wants to merge 3 commits into
18.0from
nip/18.0-account_ebics_payment_return
Draft

[MIG] accoutn_ebics_payment_return#280
ecino wants to merge 3 commits into
18.0from
nip/18.0-account_ebics_payment_return

Conversation

@ecino

@ecino ecino commented Jul 9, 2026

Copy link
Copy Markdown
Member

No description provided.

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request introduces the account_ebics_payment_return module, which enables downloading and processing PAIN.002 payment return files via the EBICS protocol in Odoo. The feedback focuses on improving the robustness of the XML parsing and recordset operations in _on_error_parse_xml_and_cancel to prevent runtime crashes, adding defensive guards when confirming payment returns, and correcting a typo in the docstring of _unlink_pain002.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

Comment on lines +86 to +158
def _on_error_parse_xml_and_cancel(self, err_message):
_logger.info("Parsing file with err: %s", err_message)
root = ET.fromstring(base64.b64decode(self.data))
ns = root.tag[1 : root.tag.index("}")]
_logger.info("PAIN002 ns: %s", ns)
po_name = root.find(
"./ns:CstmrPmtStsRpt/ns:OrgnlGrpInfAndSts/ns:OrgnlMsgId",
namespaces={"ns": ns},
).text
_logger.info("PAIN002 po_name: %s", po_name)
po_state = root.find(
"./ns:CstmrPmtStsRpt/ns:OrgnlGrpInfAndSts/ns:GrpSts", namespaces={"ns": ns}
).text
_logger.info("PAIN002 po_state: %s", po_state)
payment_order = self.env["account.payment.order"].search(
[("name", "=", po_name)]
)
_logger.info("PAIN002 payment_order: %s", payment_order)
if payment_order.state in ("generated", "uploaded"):
if po_state == "RJCT":
_logger.info(
"RJCT payment order %s with the folowing err: %s",
po_name,
err_message,
)
payment_order.action_cancel()
payment_order.message_post(body=err_message)
else:
# partially rejected only
_logger.info("Check if payment order is PART rjct")
tx = root.findall(
"./ns:CstmrPmtStsRpt/ns:OrgnlPmtInfAndSts/ns:TxInfAndSts",
namespaces={"ns": ns},
)
for t in tx:
if t.find("./ns:TxSts", namespaces={"ns": ns}).text == "RJCT":
# search for payment line
endtoend_id = t.find(
"./ns:OrgnlEndToEndId", namespaces={"ns": ns}
).text
payment_ids = payment_order.payment_ids.filtered(
lambda l: int(endtoend_id) in l.move_id.mapped("id")
)
payment_line_ids = payment_order.payment_line_ids.filtered(
lambda l: payment_ids in l.payment_ids
)
_logger.info(
f"PAIN002 payments found: {payment_ids.name} "
f"with endtoend_id: {endtoend_id}",
)

# free line with message
rsn = t.findall(
"./ns:StsRsnInf/ns:AddtlInf", namespaces={"ns": ns}
)
rsn_text = []
for r in rsn:
rsn_text.append(r.text)
rsn_txt = " ".join(rsn_text)
_logger.info(
f"PAIN002 line free: {rsn_txt} "
f"for lines {payment_line_ids}"
)
for b in payment_line_ids:
try:
b.free_line(rsn_txt)
_logger.info(f"PAIN002 line free: {rsn_txt}")
except Exception as e:
_logger.error(f"Error freeing line {b.id}: {e}")

if payment_order.state == "generated":
payment_order.generated2uploaded()
self.write({"state": "done", "note_process": err_message})

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

Critical Correctness and Robustness Issues in XML Parsing and Recordset Operations

There are several critical issues in _on_error_parse_xml_and_cancel that can lead to runtime crashes:

  1. Unsafe XML Namespace Extraction: root.tag.index("}") will raise a ValueError if the XML tag does not contain a namespace.
  2. Unsafe Node Text Access: Calling .text directly on the result of root.find(...) without checking if it is None will raise an AttributeError if the element is missing.
  3. Unsafe Integer Conversion: int(endtoend_id) will raise a ValueError if endtoend_id is not a valid integer string (e.g., if it contains non-digit characters or is empty).
  4. Potential Singleton Error: Accessing payment_ids.name will raise a ValueError: Expected singleton if payment_ids contains multiple records.
  5. Incorrect Recordset Membership Check: payment_ids in l.payment_ids is incorrect when payment_ids contains multiple records (it will always evaluate to False). Use the intersection operator & instead.
  6. Performance: Use l.move_id.ids instead of l.move_id.mapped("id") for better performance.

Here is a robust and safe implementation addressing all these issues.

    def _on_error_parse_xml_and_cancel(self, err_message):
        _logger.info("Parsing file with err: %s", err_message)
        try:
            root = ET.fromstring(base64.b64decode(self.data))
        except Exception as parse_err:
            _logger.error("Failed to parse XML: %s", parse_err)
            return

        ns = ""
        if "}" in root.tag:
            ns = root.tag[1 : root.tag.index("}")]
        _logger.info("PAIN002 ns: %s", ns)
        namespaces = {"ns": ns} if ns else {}

        po_name_node = root.find(
            "./ns:CstmrPmtStsRpt/ns:OrgnlGrpInfAndSts/ns:OrgnlMsgId",
            namespaces=namespaces,
        )
        po_name = po_name_node.text if po_name_node is not None else False
        _logger.info("PAIN002 po_name: %s", po_name)
        if not po_name:
            _logger.warning("PAIN002: Original Message ID (OrgnlMsgId) not found.")
            return

        po_state_node = root.find(
            "./ns:CstmrPmtStsRpt/ns:OrgnlGrpInfAndSts/ns:GrpSts", namespaces=namespaces
        )
        po_state = po_state_node.text if po_state_node is not None else False
        _logger.info("PAIN002 po_state: %s", po_state)

        payment_order = self.env["account.payment.order"].search(
            [("name", "=", po_name)], limit=1
        )
        _logger.info("PAIN002 payment_order: %s", payment_order)
        if payment_order and payment_order.state in ("generated", "uploaded"):
            if po_state == "RJCT":
                _logger.info(
                    "RJCT payment order %s with the following err: %s",
                    po_name,
                    err_message,
                )
                payment_order.action_cancel()
                payment_order.message_post(body=err_message)
            else:
                # partially rejected only
                _logger.info("Check if payment order is PART rjct")
                tx = root.findall(
                    "./ns:CstmrPmtStsRpt/ns:OrgnlPmtInfAndSts/ns:TxInfAndSts",
                    namespaces=namespaces,
                )
                for t in tx:
                    tx_sts_node = t.find("./ns:TxSts", namespaces=namespaces)
                    if tx_sts_node is not None and tx_sts_node.text == "RJCT":
                        # search for payment line
                        endtoend_node = t.find(
                            "./ns:OrgnlEndToEndId", namespaces=namespaces
                        )
                        endtoend_id = endtoend_node.text if endtoend_node is not None else False
                        if not endtoend_id:
                            continue

                        try:
                            move_id = int(endtoend_id)
                        except (ValueError, TypeError):
                            _logger.warning("PAIN002: OrgnlEndToEndId '%s' is not a valid integer ID.", endtoend_id)
                            continue

                        payment_ids = payment_order.payment_ids.filtered(
                            lambda l: move_id in l.move_id.ids
                        )
                        payment_line_ids = payment_order.payment_line_ids.filtered(
                            lambda l: payment_ids & l.payment_ids
                        )
                        _logger.info(
                            "PAIN002 payments found: %s with endtoend_id: %s",
                            payment_ids.mapped("name"),
                            endtoend_id,
                        )

                        # free line with message
                        rsn = t.findall(
                            "./ns:StsRsnInf/ns:AddtlInf", namespaces=namespaces
                        )
                        rsn_text = []
                        for r in rsn:
                            if r.text:
                                rsn_text.append(r.text)
                        rsn_txt = " ".join(rsn_text)
                        _logger.info(
                            "PAIN002 line free: %s for lines %s",
                            rsn_txt,
                            payment_line_ids,
                        )
                        for b in payment_line_ids:
                            try:
                                b.free_line(rsn_txt)
                                _logger.info("PAIN002 line free: %s", rsn_txt)
                            except Exception as e:
                                _logger.error("Error freeing line %s: %s", b.id, e)

                if payment_order.state == "generated":
                    payment_order.generated2uploaded()
                self.write({"state": "done", "note_process": err_message})

Comment on lines +47 to +62
payment_return = self.env["payment.return"].browse(import_result["res_id"])
# Mark the file as imported, remove binary as it should be
# attached to the statement.
_logger.info("LOG import3 '%s'", payment_return)
self.write(
{
"state": "done",
"payment_return_id": payment_return.id,
"payment_order": payment_return.payment_order_id.id,
"error_message": False,
}
)
# Automatically confirm payment returns
_logger.info("LOG import '%s'", 4)
payment_return.action_confirm()
_logger.info("[OK] import file '%s'", self.name)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Defensive Guard for Payment Return Confirmation

It is safer to check if import_result is a dictionary and contains res_id before browsing. Additionally, we should guard the call to action_confirm() to ensure it is only executed if a valid payment_return record was found.

Suggested change
payment_return = self.env["payment.return"].browse(import_result["res_id"])
# Mark the file as imported, remove binary as it should be
# attached to the statement.
_logger.info("LOG import3 '%s'", payment_return)
self.write(
{
"state": "done",
"payment_return_id": payment_return.id,
"payment_order": payment_return.payment_order_id.id,
"error_message": False,
}
)
# Automatically confirm payment returns
_logger.info("LOG import '%s'", 4)
payment_return.action_confirm()
_logger.info("[OK] import file '%s'", self.name)
res_id = import_result.get("res_id") if isinstance(import_result, dict) else False
payment_return = self.env["payment.return"].browse(res_id)
# Mark the file as imported, remove binary as it should be
# attached to the statement.
_logger.info("LOG import3 '%s'", payment_return)
self.write(
{
"state": "done",
"payment_return_id": payment_return.id,
"payment_order": payment_return.payment_order_id.id,
"error_message": False,
}
)
# Automatically confirm payment returns
_logger.info("LOG import '%s'", 4)
if payment_return:
payment_return.action_confirm()
_logger.info("[OK] import file '%s'", self.name)

Comment on lines +160 to +165
def _unlink_pain002(self):
"""
Placeholder for camt053 specific actions before removing the
EBICS data file and its related bank statements.
"""
pass

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Typo in Docstring

The docstring for _unlink_pain002 refers to camt053 instead of pain002. Update it to reflect the correct file format.

Suggested change
def _unlink_pain002(self):
"""
Placeholder for camt053 specific actions before removing the
EBICS data file and its related bank statements.
"""
pass
def _unlink_pain002(self):
"""
Placeholder for pain002 specific actions before removing the
EBICS data file and its related bank statements.
"""
pass

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants