Skip to content
Open
Show file tree
Hide file tree
Changes from 3 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
79 changes: 79 additions & 0 deletions banking/ebics/test_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
# Copyright (c) 2026, ALYF GmbH and Contributors
# See license.txt

from types import SimpleNamespace
from unittest.mock import patch

from frappe.tests.utils import FrappeTestCase

from banking.ebics.utils import UnresolvedBatchTransactionError, import_ebics_json, process_camt_document


class TestEBICSUtils(FrappeTestCase):
class EmptyBatchTransaction:
batch = True
status = None

def __len__(self):
return 0

@patch("banking.ebics.utils.get_bank_account", return_value="Test Bank Account")
@patch("banking.ebics.utils.process_camt_document")
@patch("fintech.sepa.CAMTDocument")
def test_import_batch_without_download_does_not_skip(
self, camt_document_class, process_camt_document, _get_bank_account
):
camt_document = camt_document_class.return_value
camt_document.iban = "DE89370400440532013000"
user = SimpleNamespace(
bank="Test Bank",
company="Test Company",
start_date=None,
split_batch_transactions=True,
download_batch_transactions=False,
)

import_ebics_json(user, {"camt053.xml": b"<Document />"})

process_camt_document.assert_called_once_with(
camt_document,
"Test Bank Account",
"Test Company",
None,
True,
False,
)

@patch("banking.ebics.utils.create_sepa_bank_transaction")
@patch("banking.ebics.utils.get_transaction_id", return_value="batch-id")
def test_process_unresolved_batch_without_download_as_single_transaction(
self, _get_transaction_id, create_sepa_bank_transaction
Comment thread
barredterra marked this conversation as resolved.
):
transaction = self.EmptyBatchTransaction()

process_camt_document(
[transaction],
"Test Bank Account",
"Test Company",
split_batch_transactions=True,
skip_unresolved_batch_transactions=False,
)

create_sepa_bank_transaction.assert_called_once_with(
"Test Bank Account",
"Test Company",
transaction,
transaction_id="batch-id",
start_date=None,
)

@patch("banking.ebics.utils.get_transaction_id", return_value="batch-id")
def test_process_unresolved_batch_with_download_is_skipped(self, _get_transaction_id):
with self.assertRaises(UnresolvedBatchTransactionError):
process_camt_document(
[self.EmptyBatchTransaction()],
"Test Bank Account",
"Test Company",
split_batch_transactions=True,
skip_unresolved_batch_transactions=True,
)
49 changes: 27 additions & 22 deletions banking/ebics/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -256,6 +256,7 @@ def import_ebics_json(user: EBICSUser, main_data: dict, batch_data: dict | None
user.company,
user.start_date,
user.split_batch_transactions,
user.download_batch_transactions,
Comment thread
barredterra marked this conversation as resolved.
Outdated
)


Expand Down Expand Up @@ -292,6 +293,7 @@ def process_camt_document(
company: str | None = None,
earliest_date: date | None = None,
split_batch_transactions: bool = False,
skip_unresolved_batch_transactions: bool = True,
):
if not company:
company = frappe.db.get_value("Bank Account", bank_account, "company")
Expand All @@ -307,28 +309,31 @@ def process_camt_document(
# Split batch transactions into sub-transactions, based on info from camt.054.

if len(transaction) == 0:
# camt.054 might become available at a later time than camt.053.
# In this case, we want to block the processing of camt.053 until camt.054 is available.
raise UnresolvedBatchTransactionError()

for sub_transaction_index, sub_transaction in enumerate(transaction):
sub_transaction_id = get_transaction_id(sub_transaction, sub_transaction_index)
create_sepa_bank_transaction(
bank_account,
company,
sub_transaction,
transaction_id=transaction_id,
subtransaction_id=sub_transaction_id,
start_date=earliest_date,
)
else:
create_sepa_bank_transaction(
bank_account,
company,
transaction,
transaction_id=transaction_id,
start_date=earliest_date,
)
if skip_unresolved_batch_transactions:
# camt.054 might become available at a later time than camt.053.
# In this case, we want to block the processing of camt.053 until camt.054 is available.
raise UnresolvedBatchTransactionError()

else:
for sub_transaction_index, sub_transaction in enumerate(transaction):
sub_transaction_id = get_transaction_id(sub_transaction, sub_transaction_index)
create_sepa_bank_transaction(
bank_account,
company,
sub_transaction,
transaction_id=transaction_id,
subtransaction_id=sub_transaction_id,
start_date=earliest_date,
)
continue

create_sepa_bank_transaction(
bank_account,
company,
transaction,
transaction_id=transaction_id,
start_date=earliest_date,
)


def create_sepa_bank_transaction(
Expand Down
Loading