import base64
from datetime import datetime, timedelta
from lxml import etree
from odoo import Command, fields, tools
from odoo.exceptions import UserError
from odoo.tests import freeze_time, patch, tagged
from odoo.addons.account.tests.common import AccountTestInvoicingCommon
from odoo.addons.base.tests.test_ir_cron import CronMixinCase
from odoo.addons.l10n_pl_edi.models.account_move import AccountMove
from odoo.addons.l10n_pl_edi.tools.ksef_api_service import KsefApiService
def attachment_to_dict(attachment):
return {'name': attachment.name, 'raw': attachment.raw}
@tagged('post_install', '-at_install', 'post_install_l10n')
class TestL10nPlEdi(AccountTestInvoicingCommon, CronMixinCase):
@classmethod
@AccountTestInvoicingCommon.setup_country('pl')
@freeze_time('2026-01-23')
def setUpClass(cls):
super().setUpClass()
cls.country_pl = cls.env.ref('base.pl')
cls.company = cls.company_data['company']
cls.company.write({
'country_id': cls.country_pl.id,
'vat': 'PL1234567883',
'street': 'Test Street 1',
'city': 'Warsaw',
'zip': '00-001',
})
cls.company_2 = cls.setup_other_company(
name='PL2 Company',
vat='PL1111111111',
street='Other Street 2',
city='Krakow',
zip='30-001',
)['company']
cls.partner_pl = cls.env['res.partner'].create({
'name': 'Test Customer PL',
'is_company': True,
'country_id': cls.country_pl.id,
'vat': 'PL1111111111',
'street': 'Partner St. 5',
'city': 'Krakow',
'zip': '30-001',
})
cls.tax_23 = cls.company_data['default_tax_sale'].copy({
'amount': 23.0,
'name': 'VAT 23%',
'amount_type': 'percent',
})
cls.product_a.taxes_id = cls.tax_23
cls.cash_journal = cls.company_data['default_journal_cash']
cls.cash_journal.inbound_payment_method_line_ids.payment_account_id = cls.cash_journal.default_account_id.id
cls.env['ir.config_parameter'].sudo().set_str('l10n_pl_edi_ksef.mode', 'test')
def read_certificate_file(filename, b64=False):
path = f'l10n_pl_edi/tests/certificate/{filename}'
with tools.file_open(path, mode='rb') as fd:
content = fd.read()
return base64.b64encode(content) if b64 else content
# This Certificate will NOT work for the current test VAT with KSEF (not even in test mode),
# AND will even expire, but it's useful to have something to test with when we mock
# the API calls
key_filename, cert_filename = 'l10n_pl_edi_test.key', 'l10n_pl_edi_test.pem'
key = cls.env['certificate.key'].create(dict(
company_id=cls.company.id,
name=key_filename,
content=read_certificate_file(key_filename, b64=True),
password='Qwertyuiop@12345',
))
cert = cls.env['certificate.certificate'].create(dict(
company_id=cls.company.id,
name=cert_filename,
content=read_certificate_file(cert_filename, b64=True),
private_key_id=key.id,
))
cls.company.sudo().write({
'l10n_pl_edi_register': True,
'l10n_pl_edi_certificate': cert.id,
'l10n_pl_edi_access_token': "aa33ccee",
'l10n_pl_edi_refresh_token': "bb44ddff",
})
cls.company_2.sudo().write({
'l10n_pl_edi_register': True,
'l10n_pl_edi_certificate': cert.id,
'l10n_pl_edi_access_token': "aa33ccee",
'l10n_pl_edi_refresh_token': "bb44ddff",
})
cls.standard_invoice = cls.env['account.move'].create({
'move_type': 'out_invoice',
'partner_id': cls.partner_pl.id,
'invoice_date': fields.Date.today(),
'invoice_line_ids': [
Command.create({
'product_id': cls.product_a.id,
'quantity': 1,
'price_unit': 1000.0,
})
],
'ref': '12345',
})
def _get_xml_value(self, xml_content, xpath):
"""Helper to parse XML and return text of a specific node."""
nodes = self._get_xml_nodes(xml_content, xpath)
if nodes:
return nodes[0].text
return ""
def _get_xml_nodes(self, xml_content, xpath):
"""Helper to return a list of nodes."""
root = etree.fromstring(xml_content)
ns = {'ns': 'http://crd.gov.pl/wzor/2025/06/25/13775/'}
return root.xpath(xpath, namespaces=ns)
def _assert_export_invoice(self, invoice, filename):
path = f'l10n_pl_edi/tests/export_xmls/{filename}'
with tools.file_open(path, mode='rb') as fd:
expected_tree = etree.fromstring(fd.read())
xml = invoice._l10n_pl_edi_render_xml()
invoice_etree = etree.fromstring(xml)
try:
self.assertXmlTreeEqual(invoice_etree, expected_tree)
except AssertionError as ae:
ae.args = (ae.args[0] + f"\nFile used for comparison: {filename}", )
raise
def _assert_import_invoice(self, filename, expected_values):
path = f'l10n_pl_edi/tests/import_xmls/{filename}'
with tools.file_open(path, mode='rb') as fd:
invoice = self.env['account.move'].create(self.env['account.move'].l10n_pl_edi_get_ksef_bill_vals_from_xml(fd.read()))
try:
self.assertRecordValues(invoice.invoice_line_ids, expected_values)
except AssertionError as ae:
ae.args = (ae.args[0] + f"\nFile used for comparison: {filename}", )
raise
@freeze_time('2026-01-23')
def test_ksef_fa3_standard_vat(self):
"""
Standard VAT Invoice.
This test verifies that a regular Odoo invoice (not a down payment or correction)
generates a KSeF XML with the invoice type VAT.
It simulates a simple sale of a product.
"""
self.standard_invoice.action_post()
self._assert_export_invoice(self.standard_invoice, "standert_fa3_format.xml")
@freeze_time('2026-01-23')
def test_p6_delivery_date_differs_from_invoice_date(self):
"""
P_6 (delivery date) must be emitted when the delivery date differs from the
invoice issue date (invoice_date), regardless of the accounting date (date).
"""
invoice = self._create_invoice(
partner_id=self.partner_pl,
invoice_date='2025-05-27',
date='2026-05-04',
delivery_date='2026-05-04',
post=True
)
xml = invoice._l10n_pl_edi_render_xml()
self.assertEqual(
self._get_xml_value(xml, "//ns:Fa/ns:P_6"),
'2026-05-04',
)
@freeze_time('2026-01-23')
def test_scenario_correction_standard(self):
"""
Correction of a Standard Invoice (KOR).
This test verifies that creating a Credit Note (reversal) for a standard invoice
generates a KSeF XML with invoice type KOR.
"""
# Create Invoice
invoice = self.env['account.move'].create({
'move_type': 'out_invoice',
'partner_id': self.partner_pl.id,
'invoice_date': fields.Date.today(),
'invoice_line_ids': [Command.create({'product_id': self.product_a.id, 'price_unit': 1000.0})],
})
invoice.action_post()
reversal_wizard = self.env['account.move.reversal'].create({
'reason': 'Correction Test',
'journal_id': invoice.journal_id.id,
'move_ids': invoice.ids,
})
reversal_wizard.refund_moves()
credit_note = invoice.reversal_move_ids
credit_note.action_post()
self._assert_export_invoice(credit_note, 'standerd_fa3_credit_note.xml')
@freeze_time('2026-01-23')
def test_ksef_fa3_reverse_charge(self):
invoice_lines = self._prepare_invoice_line(product_id=self.product.id, quantity=1,
price_unit=1000.0, tax_ids=[Command.set(self.env['account.chart.template'].ref('vs_stal').ids)])
invoice = self._create_invoice(invoice_date=fields.Date.today(), partner_id=self.partner_pl,
invoice_line_ids=[invoice_lines], post=True)
self._assert_export_invoice(invoice, 'standard_fa3_format_invoice_reverse_charge.xml')
@freeze_time('2026-01-23')
def test_payment_logic_partial_mixed_methods(self):
"""
Test the block for a Partially Paid invoice with mixed methods.
We expect:
- ZnacznikZaplatyCzesciowej = 1
- Two ZaplataCzesciowa blocks.
- FormaPlatnosci = 1 for Cash.
- FormaPlatnosci = 6 for Bank.
"""
invoice = self.env['account.move'].create({
'move_type': 'out_invoice',
'partner_id': self.partner_pl.id,
'invoice_date': fields.Date.today(),
'currency_id': self.env.ref('base.PLN').id,
'invoice_line_ids': [Command.create({
'product_id': self.product_a.id,
'quantity': 1,
'price_unit': 1000.0,
})],
})
invoice.action_post()
self.env['account.payment.register'].create({
'amount': 300.0,
'journal_id': self.cash_journal.id,
'payment_date': fields.Date.today(),
'line_ids': invoice.line_ids.filtered(lambda line: line.display_type == 'payment_term'),
})._create_payments()
self.env['account.payment.register'].create({
'amount': 400.0,
'journal_id': self.cash_journal.id,
'payment_date': fields.Date.today(),
'line_ids': invoice.line_ids.filtered(lambda line: line.display_type == 'payment_term'),
})._create_payments()
self.assertEqual(invoice.payment_state, 'partial')
self._assert_export_invoice(invoice, 'partial_paid_invoice_bank_cash.xml')
@freeze_time('2026-01-23')
def test_payment_logic_fully_paid(self):
"""
Test the block for a Fully Paid invoice.
We expect:
- Zaplacono = 1
- ZnacznikZaplatyCzesciowej = 2 (Not partial, because it is fully paid)
- DataZaplaty present
"""
invoice = self.env['account.move'].create({
'move_type': 'out_invoice',
'partner_id': self.partner_pl.id,
'invoice_date': fields.Date.today(),
'invoice_line_ids': [Command.create({'product_id': self.product_a.id, 'price_unit': 100.0})],
})
invoice.action_post()
self.env['account.payment.register'].create({
'journal_id': self.cash_journal.id,
'line_ids': invoice.line_ids.filtered(lambda line: line.display_type == 'payment_term'),
})._create_payments()
self.assertEqual(invoice.payment_state, 'paid')
self._assert_export_invoice(invoice, 'full_paid_invoice.xml')
@freeze_time('2026-01-23')
def test_payment_logic_partial_then_full_payment(self):
"""
Test the block when an invoice is paid in installments (Partial -> Full).
Scenario:
1. Create Invoice for 1000 PLN.
2. Pay 400 PLN (Status becomes Partial).
3. Pay remaining 600 PLN (Status becomes Paid).
Expectations:
- ZnacznikZaplatyCzesciowej = 2 (It is fully paid, so flag is 2/No).
- ZaplataCzesciowa nodes should be present (listing the 2 payments).
- Zaplacono should NOT be present (based on your template logic for multi-payment).
"""
# 1. Create Invoice
invoice = self.env['account.move'].create({
'move_type': 'out_invoice',
'partner_id': self.partner_pl.id,
'invoice_date': fields.Date.today(),
'currency_id': self.env.ref('base.PLN').id,
'invoice_line_ids': [Command.create({
'product_id': self.product_a.id,
'quantity': 1,
'price_unit': 1000.0,
'tax_ids': [],
})],
})
invoice.action_post()
# 2. Register First Partial Payment (400 PLN)
self.env['account.payment.register'].create({
'amount': 400.0,
'journal_id': self.cash_journal.id,
'payment_date': fields.Date.today(),
'line_ids': invoice.line_ids.filtered(lambda line: line.display_type == 'payment_term'),
})._create_payments()
self.assertEqual(invoice.payment_state, 'partial')
# 3. Register Remaining 600 PLN Payment
self.env['account.payment.register'].create({
'amount': 600.0,
'journal_id': self.cash_journal.id,
'payment_date': fields.Date.today(),
'line_ids': invoice.line_ids.filtered(lambda line: line.display_type == 'payment_term'),
})._create_payments()
self.assertEqual(invoice.payment_state, 'paid')
self.assertEqual(len(invoice._get_reconciled_payments()), 2)
# 4. Render XML
xml = invoice._l10n_pl_edi_render_xml()
# Expectation: ZnacznikZaplatyCzesciowej is 2 because invoice is fully paid
self.assertEqual(
self._get_xml_value(xml, "//ns:Platnosc/ns:ZnacznikZaplatyCzesciowej"),
'2',
"ZnacznikZaplatyCzesciowej should be 2 when fully paid (even with multiple payments)"
)
# Expectation: ZaplataCzesciowa nodes SHOULD be present for both payments
payment_nodes = self._get_xml_nodes(xml, "//ns:Platnosc/ns:ZaplataCzesciowa")
self.assertEqual(len(payment_nodes), 2, "Should list history of both payments")
# Optional: Verify amounts in the history
amounts = sorted([n.find('ns:KwotaZaplatyCzesciowej', namespaces={'ns': 'http://crd.gov.pl/wzor/2025/06/25/13775/'}).text for n in payment_nodes])
self.assertEqual(amounts, ['400.00', '600.00'])
def test_payment_bank_account_details(self):
"""
Test that RachunekBankowy is generated when a partner_bank_id is set on the invoice.
"""
# Create a Bank Account for the Company
bank_acc = self.env['res.partner.bank'].create({
'account_number': '12 3456 7890 0000 0000 1234 5678',
'partner_id': self.partner_pl.id,
'bank_name': "Test Bank PL",
'allow_out_payment': True,
})
invoice = self.env['account.move'].create({
'move_type': 'out_invoice',
'partner_id': self.partner_pl.id,
'invoice_date': fields.Date.today(),
'partner_bank_id': bank_acc.id,
'invoice_line_ids': [Command.create({'product_id': self.product_a.id, 'price_unit': 100.0})],
})
invoice.action_post()
self.env['account.payment.register'].create({
'amount': 10.0,
'journal_id': self.cash_journal.id,
'line_ids': invoice.line_ids.filtered(lambda line: line.display_type == 'payment_term'),
})._create_payments()
xml = invoice._l10n_pl_edi_render_xml()
expected_acc = '12345678900000000012345678'
self.assertEqual(self._get_xml_value(xml, "//ns:Platnosc/ns:RachunekBankowy/ns:NrRB"), expected_acc)
self.assertEqual(self._get_xml_value(xml, "//ns:Platnosc/ns:RachunekBankowy/ns:NazwaBanku"), 'Test Bank PL')
@freeze_time('2026-01-23')
def test_payment_terms_structure(self):
"""
Test the block logic.
Scenario:
1. Create a custom Payment Term: "45 Days after End of Month".
2. Create an Invoice using this term.
3. Verify the XML output contains:
- Termin: The calculated due date.
- Ilosc: 45
- Jednostka: Dni
- ZdarzeniePoczatkowe: Koniec miesiąca
"""
# 1. Create Custom Payment Term (45 Days After End of Month)
pay_term = self.env['account.payment.term'].create({
'name': '45 Days EOM',
'line_ids': [
Command.create({
'value': 'percent',
'value_amount': 100.0,
'nb_days': 45,
'delay_type': 'days_after_end_of_month',
})
]
})
# 2. Create Invoice with this Payment Term
invoice = self.env['account.move'].create({
'move_type': 'out_invoice',
'partner_id': self.partner_pl.id,
'invoice_date': fields.Date.today(),
'invoice_payment_term_id': pay_term.id,
'invoice_line_ids': [Command.create({
'product_id': self.product_a.id,
'price_unit': 100.0
})],
})
invoice.action_post()
self.env['account.payment.register'].create({
'amount': 100.0,
'journal_id': self.cash_journal.id,
'line_ids': invoice.line_ids.filtered(lambda line: line.display_type == 'payment_term'),
})._create_payments()
self.assertEqual(invoice.payment_state, 'partial')
# 3. Render XML
xml = invoice._l10n_pl_edi_render_xml()
expected_date = str(invoice.invoice_date_due)
self.assertEqual(
self._get_xml_value(xml, "//ns:Platnosc/ns:TerminPlatnosci/ns:Termin"),
expected_date,
"Termin should match the invoice due date"
)
# Check TerminOpis (Structured Description)
# Ilosc (Quantity)
self.assertEqual(
self._get_xml_value(xml, "//ns:Platnosc/ns:TerminPlatnosci/ns:TerminOpis/ns:Ilosc"),
'45',
"Ilosc should be 45"
)
# Jednostka (Unit)
self.assertEqual(
self._get_xml_value(xml, "//ns:Platnosc/ns:TerminPlatnosci/ns:TerminOpis/ns:Jednostka"),
'Dni',
"Jednostka should be Dni"
)
@freeze_time('2026-01-23')
def test_invoice_in_foreign_currency(self):
self.env.ref('base.EUR').active = True
invoice = self.env['account.move'].create({
'move_type': 'out_invoice',
'partner_id': self.partner_pl.id,
'invoice_date': fields.Date.today(),
'currency_id': self.env.ref('base.EUR').id,
'invoice_line_ids': [Command.create({
'product_id': self.product_a.id,
'quantity': 1,
'price_unit': 10.0,
})],
'invoice_currency_rate': '2',
})
invoice.action_post()
self._assert_export_invoice(invoice, 'fa3_invoice_foreign_currency.xml')
@freeze_time('2026-01-23')
def test_prefiks_podatnika_for_eu_transactions(self):
"""
Verification that PrefiksPodatnika ('PL') is included in Podmiot1 for invoices
documenting intra-Community supply of goods (K_21), EU B2B services (K_12),
or simplified triangular transactions (Triangular Sale), as required by
Art. 97 sec. 10 items 2-3 and Art. 136 sec. 1 item 3 of the Polish VAT Act.
"""
invoices = self.env['account.move'].create([
{
'move_type': 'out_invoice',
'partner_id': self.partner_pl.id,
'invoice_date': fields.Date.today(),
'invoice_line_ids': [Command.create({
'product_id': self.product_a.id,
'price_unit': 100.0,
'tax_ids': self.env['account.chart.template'].ref(tax_ref).ids,
})],
} for tax_ref in ['vs_unia', 'vs_dostu', 'vs_unia_triangular']
])
invoices.action_post()
for invoice in invoices:
xml = invoice._l10n_pl_edi_render_xml()
self.assertEqual(
self._get_xml_value(xml, "//ns:Podmiot1/ns:PrefiksPodatnika"),
'PL',
)
@freeze_time('2026-01-23')
def test_ksef_fa3_eu_service_b2b_includes_p13_9(self):
"""
EU B2B service invoices tagged with K_12 must include P_13_9 as the net
amount of the supplied service.
"""
service_tax = self.env['account.chart.template'].ref('vs_dostu')
service_product = self._create_product(name='EU Service', type='service')
invoice_line = self._prepare_invoice_line(product_id=service_product.id, quantity=1, price_unit=1000.0, tax_ids=[(6, 0, [service_tax.id])])
invoice = self._create_invoice(invoice_line_ids=[invoice_line], partner_id=self.partner_pl.id, post=True)
xml = invoice._l10n_pl_edi_render_xml()
self.assertEqual(self._get_xml_value(xml, "//ns:Fa/ns:P_13_9"), '1000.00')
@freeze_time('2026-01-23')
def test_ksef_fa3_non_eu_service_generates_p_13_8(self):
"""
Non-EU service invoices (K_11) must include P_13_8 as the net amount of the supplied service.
"""
non_eu_service_product = self._create_product(name='Non-EU Service', type='service')
invoice_line = self._prepare_invoice_line(product_id=non_eu_service_product.id, quantity=1,
price_unit=2000.0, tax_ids=[Command.set(self.env['account.chart.template'].ref('vs_ekspu').ids)])
invoice = self._create_invoice(invoice_line_ids=[invoice_line], partner_id=self.partner_pl.id, post=True)
xml = invoice._l10n_pl_edi_render_xml()
self.assertEqual(self._get_xml_value(xml, "//ns:Fa/ns:P_13_8"), '2000.00')
@freeze_time('2026-01-23')
def test_scenario_correction_values_are_negative(self):
"""
Verification of Negative Values for Corrections (Difference Method).
This test ensures that when a Credit Note (KOR) is generated:
1. Quantity (P_8B) is NEGATIVE.
2. Net Amount (P_11) is NEGATIVE.
3. Unit Price (P_9A) is POSITIVE.
"""
invoice = self.env['account.move'].create({
'move_type': 'out_invoice',
'partner_id': self.partner_pl.id,
'invoice_date': fields.Date.today(),
'currency_id': self.env.ref('base.PLN').id,
'invoice_line_ids': [Command.create({
'product_id': self.product_a.id,
'quantity': 10,
'price_unit': 100.0,
})],
})
invoice.action_post()
reversal_wizard = self.env['account.move.reversal'].create({
'reason': 'Return of goods',
'journal_id': invoice.journal_id.id,
'move_ids': invoice.ids,
})
reversal_wizard.refund_moves()
credit_note = invoice.reversal_move_ids
credit_note.action_post()
xml = credit_note._l10n_pl_edi_render_xml()
self.assertEqual(self._get_xml_value(xml, "//ns:RodzajFaktury"), 'KOR')
p_8b = self._get_xml_value(xml, "//ns:Fa/ns:FaWiersz/ns:P_8B")
self.assertEqual(float(p_8b), -10.0, "Quantity (P_8B) must be negative for corrections")
p_11 = self._get_xml_value(xml, "//ns:Fa/ns:FaWiersz/ns:P_11")
self.assertEqual(float(p_11), -1000.0, "Net Amount (P_11) must be negative for corrections")
p_9a = self._get_xml_value(xml, "//ns:Fa/ns:FaWiersz/ns:P_9A")
self.assertEqual(float(p_9a), 100.0, "Unit Price (P_9A) must remain positive")
def l10n_pl_edi_generate_attachments(self, invoices, from_cron=False):
moves_data = {
invoice: self.env['account.move.send']._get_default_sending_settings(invoice, from_cron=from_cron)
for invoice in invoices
}
with patch('odoo.addons.l10n_pl_edi.models.account_move_send.AccountMoveSend._call_web_service_before_invoice_pdf_render'):
self.env['account.move.send']._generate_invoice_documents(moves_data)
def test_l10n_pl_edi_send_success(self):
invoice = self.standard_invoice
invoice.action_post()
with (
patch.object(KsefApiService, 'open_ksef_session') as mock_open_session,
patch.object(KsefApiService, 'send_invoice', return_value={'referenceNumber': '999999'}) as mock_send,
patch.object(KsefApiService, 'get_invoice_status', return_value={
'ksefNumber': '7492091229-20260210-0700A043714A-5E',
'status': {'code': 200},
})
):
wizard = self.env['account.move.send.wizard'].with_company(self.company).create({
'move_id': invoice.id,
'extra_edi_checkboxes': {'pl_ksef': {'checked': True}}
})
wizard.action_send_and_print()
self.assertEqual(mock_open_session.call_count, 1)
self.assertEqual(mock_send.call_count, 1)
self.assertEqual(invoice.l10n_pl_edi_status, 'accepted')
self.assertEqual(invoice.l10n_pl_edi_session_id, invoice.company_id.l10n_pl_edi_session_id)
self.assertEqual(invoice.l10n_pl_edi_ref, '999999')
self.assertEqual(invoice.l10n_pl_edi_number, '7492091229-20260210-0700A043714A-5E')
self.assertEqual(invoice.l10n_pl_edi_attachment_id.name, 'FA3-INV_2026_00001.xml')
def test_l10n_pl_edi_send_api_error(self):
invoice = self.standard_invoice
invoice.action_post()
def send_invoice_raise(xml_content):
raise Exception("turlututu")
with (
patch.object(KsefApiService, 'open_ksef_session') as mock_open_session,
patch.object(KsefApiService, 'send_invoice', side_effect=send_invoice_raise) as mock_send,
self.assertRaises(UserError),
):
wizard = self.env['account.move.send.wizard'].with_company(self.company).create({
'move_id': invoice.id,
'extra_edi_checkboxes': {'pl_ksef': {'checked': True}}
})
wizard.action_send_and_print()
self.assertEqual(mock_open_session.call_count, 1)
self.assertEqual(mock_send.call_count, 1)
self.assertEqual(invoice.l10n_pl_edi_status, False)
self.assertEqual(invoice.l10n_pl_edi_session_id, False)
self.assertEqual(invoice.l10n_pl_edi_ref, False)
self.assertEqual(invoice.l10n_pl_edi_attachment_id.name, False)
def test_l10n_pl_edi_download_bill_success(self):
def query_invoice_metadata(query_criteria, page_size=100, page_offset=0):
return {
'hasMore': False,
'invoices': [
{
'acquisitionDate': '2026-02-10T10:50:29.348439+00:00',
'buyer': {'identifier': {'type': 'Nip', 'value': '5795955811'}, 'name': 'PL Company'},
'currency': 'PLN',
'formCode': {'schemaVersion': '1-0E', 'systemCode': 'FA (3)', 'value': 'FA'},
'hasAttachment': False,
'invoiceHash': 'DOFApZsfUkl3BLgW1nd7frNq4IVHvYoXHEudpyCFbpg=',
'invoiceNumber': 'FV/2026/00001-demo-test-005',
'invoiceType': 'Vat',
'invoicingDate': '2026-02-10T10:50:29.189345+00:00',
'invoicingMode': 'Online',
'isSelfInvoicing': False,
'issueDate': '2026-02-10',
'ksefNumber': '7492091229-20260210-0700A043714A-5E',
'permanentStorageDate': '2026-02-10T10:50:30.40494+00:00',
'seller': {'name': 'HADRON FOR BUSINESS SP Z O O', 'nip': '7492091229'},
},
],
}
def get_invoice_by_ksef_number(ksef_number):
path = 'l10n_pl_edi/tests/export_xmls/fa3_bill.xml'
with tools.file_open(path, mode='rb') as file:
return {'xml_content': file.read()}
with (
patch.object(KsefApiService, 'query_invoice_metadata', side_effect=query_invoice_metadata),
patch.object(KsefApiService, 'get_invoice_by_ksef_number', side_effect=get_invoice_by_ksef_number),
):
self.env['account.move'].with_company(self.company)._l10n_pl_edi_download_bills_from_ksef()
created_move = self.env['account.move'].search([('l10n_pl_edi_number', '=', '7492091229-20260210-0700A043714A-5E')])
self.assertTrue(created_move)
self.assertEqual(created_move.partner_id.vat, 'PL7492091229')
self.assertEqual(len(created_move), 1)
self.assertRecordValues(
created_move, [
{
'state': 'draft',
'move_type': 'in_invoice',
'invoice_date': fields.Date.to_date('2026-02-10'),
'invoice_date_due': fields.Date.to_date('2026-02-10'),
'ref': 'FV/2026/00001-demo-test-005',
'currency_id': self.env['res.currency'].search([('name', '=', 'PLN')]).id,
},
],
)
self.assertRecordValues(
created_move.invoice_line_ids, [
{
'name': "[FURN_0006] Podstawka pod monitor",
'quantity': 1.0,
'price_unit': 3.19,
},
{
'name': "[FOOD_0001] Chleb pszenny",
'quantity': 2.5,
'price_unit': 5.00,
},
{
'name': "[BOOK_0001] Podręcznik szkolny",
'quantity': 4.0,
'price_unit': 5.00,
},
],
)
self.assertEqual(
created_move.invoice_line_ids.tax_ids.ids,
self.env['account.tax'].search(
[
('name', 'in', ('23% G', '8%', '5%')),
('type_tax_use', '=', 'purchase'),
*self.env['account.tax']._check_company_domain(self.company),
],
).ids,
)
created_move_attachment = self.env['ir.attachment'].search([
('res_model', '=', 'account.move'),
('res_id', '=', created_move.id),
], limit=1)
self.assertTrue(created_move_attachment)
with tools.file_open('l10n_pl_edi/tests/export_xmls/fa3_bill.xml', mode='rb') as file:
self.assertEqual(created_move_attachment.raw, file.read())
def test_l10n_pl_edi_download_bill_retry_after(self):
"""Test that when a rate limit error occurs the progress is preserved and the cron is rescheduled."""
def query_invoice_metadata(query_criteria, page_size=100, page_offset=0):
return {
'hasMore': False,
'invoices': [
{
'ksefNumber': 'KSEF-BILL-001',
},
{
'ksefNumber': 'KSEF-BILL-002',
},
],
}
call_count = 0
def get_invoice_by_ksef_number(ksef_number):
nonlocal call_count
call_count += 1
if call_count == 1:
path = 'l10n_pl_edi/tests/export_xmls/fa3_bill.xml'
with tools.file_open(path, mode='rb') as file:
return {'xml_content': file.read()}
return {'error': {'retry_after': 120, 'message': 'Too Many Requests'}}
start = fields.Datetime.now()
with (
patch.object(KsefApiService, 'query_invoice_metadata', side_effect=query_invoice_metadata),
patch.object(KsefApiService, 'get_invoice_by_ksef_number', side_effect=get_invoice_by_ksef_number),
self.capture_triggers() as capt,
):
cron_runs_before = len(capt.records)
self.env['account.move'].with_company(self.company)._l10n_pl_edi_download_bills_from_ksef()
bill_1 = self.env['account.move'].search([('l10n_pl_edi_status', '=', 'fetched')])
self.assertTrue(bill_1)
bill_1_attachment = self.env['ir.attachment'].search([
('res_model', '=', 'account.move'),
('res_id', '=', bill_1.id),
], limit=1)
self.assertTrue(bill_1_attachment)
with tools.file_open('l10n_pl_edi/tests/export_xmls/fa3_bill.xml', mode='rb') as file:
self.assertEqual(bill_1_attachment.raw, file.read())
bill_2 = self.env['account.move'].search([('l10n_pl_edi_status', '=', 'fetch_ready')])
self.assertTrue(bill_2)
self.assertEqual(len(capt.records), cron_runs_before + 1)
self.assertGreaterEqual(capt.records[-1].call_at, start + timedelta(seconds=120))
self.assertLessEqual(capt.records[-1].call_at, start + timedelta(seconds=240))
def test_l10n_pl_edi_download_problematic_bill_do_not_stop_others(self):
"""Test that when an error occurs on a single bill, the rest don't get stuck."""
def query_invoice_metadata(query_criteria, page_size=100, page_offset=0):
return {
'hasMore': False,
'invoices': [{'ksefNumber': 'KSEF-BILL-001'}, {'ksefNumber': 'KSEF-BILL-002'}],
}
def get_invoice_by_ksef_number(ksef_number):
return {'xml_content': b'foo'}
call_count = 0
def l10n_pl_edi_get_ksef_bill_vals_from_xml(xml_content):
nonlocal call_count
if call_count == 0:
call_count = 1
return {'currency_id': None} # Should raise a NOT NULL db error
return {'currency_id': self.env.company.currency_id.id}
with (
patch.object(KsefApiService, 'query_invoice_metadata', side_effect=query_invoice_metadata),
patch.object(KsefApiService, 'get_invoice_by_ksef_number', side_effect=get_invoice_by_ksef_number),
patch.object(AccountMove, 'l10n_pl_edi_get_ksef_bill_vals_from_xml', side_effect=l10n_pl_edi_get_ksef_bill_vals_from_xml),
):
self.env['account.move'].with_company(self.company)._l10n_pl_edi_download_bills_from_ksef()
bills = self.env['account.move'].search([('l10n_pl_edi_number', 'in', ('KSEF-BILL-001', 'KSEF-BILL-002'))])
bills_attachments = self.env['ir.attachment'].search([
('res_model', '=', 'account.move'),
('res_id', 'in', bills.ids),
])
self.assertEqual(len(bills), 2)
self.assertEqual(len(bills_attachments), 2)
self.assertTrue(bills.filtered(lambda b: b.l10n_pl_edi_status == 'fetch_failed'))
self.assertTrue(bills.filtered(lambda b: b.l10n_pl_edi_status == 'fetched'))
def test_import_invoice_with_net_and_gross_unit_price(self):
self._assert_import_invoice('invoice_with_net_and_gross_unit_price.xml', [
{
'name': '[FURN_0006] Podstawka pod monitor',
'price_unit': 3.19,
'price_total': 3.92,
'tax_ids': self.env['account.chart.template'].ref('vz_kraj_23').ids,
},
{
'name': '[FOOD_0001] Chleb pszenny',
'price_unit': 5.0,
'price_total': 13.5,
'tax_ids': self.env['account.chart.template'].ref('vz_kraj_8').ids,
},
{
'name': '[BOOK_0001] Podręcznik szkolny',
'price_unit': 5.0,
'price_total': 21.0,
'tax_ids': self.env['account.chart.template'].ref('vz_kraj_5').ids,
},
])
def test_import_invoice_with_zero_value_line_missing_unit_prices(self):
self._assert_import_invoice('invoice_with_zero_value_line_missing_unit_prices.xml', [
{
'name': '[FURN_0006] Podstawka pod monitor',
'price_unit': 3.19,
'price_total': 3.92,
'tax_ids': self.env['account.chart.template'].ref('vz_kraj_23').ids,
},
{
'name': '[ZERO] Transport',
'price_unit': 0.0,
'price_total': 0.0,
'tax_ids': self.env['account.chart.template'].ref('vz_kraj_zw').ids,
},
])
def test_import_invoice_from_foreign_country_retrieve_correct_partner(self):
partner = self.env["res.partner"].create({
'name': "LU Company",
'vat': "PL7492091229",
'country_id': self.env.ref('base.lu').id,
})
self._assert_import_invoice('invoice_from_foreign_country.xml', [{
'partner_id': partner.id,
}])
def test_import_invoice_from_foreign_country_partner_correctly_created(self):
path = 'l10n_pl_edi/tests/import_xmls/invoice_from_foreign_country.xml'
with tools.file_open(path, mode='rb') as fd:
content = fd.read()
invoice_data = self.env['account.move'].l10n_pl_edi_get_ksef_bill_vals_from_xml(content)
invoice = self.env['account.move'].create(invoice_data)
self.assertEqual(invoice.partner_id.name, "LU Company")
self.assertIn("7492091229", invoice.partner_id.vat)
@freeze_time('2026-01-23')
def test_ksef_export_vat_without_country_code(self):
""" Ensure VAT numbers do not include the country code in NrVatUE/NrID fields """
# non-Polish VAT
foreign_partner = self.partner_a.copy({'vat': 'GB298357641'})
invoice = self._create_invoice(partner_id=foreign_partner.id, post=True)
xml = invoice._l10n_pl_edi_render_xml()
self.assertEqual(self._get_xml_value(xml, "//ns:Podmiot2/ns:DaneIdentyfikacyjne/ns:NrID"), '298357641')
# Polish VAT
invoice = self._create_invoice(partner_id=self.partner_pl.id, post=True)
xml = invoice._l10n_pl_edi_render_xml()
self.assertEqual(self._get_xml_value(xml, "//ns:Podmiot2/ns:DaneIdentyfikacyjne/ns:NIP"), '1111111111')
def test_cron_creates_empty_move_on_parsing_error(self):
""" Create empty journal entry for malformed XML """
def query_invoice_metadata(query_criteria, page_size=100, page_offset=0):
return {
'hasMore': False,
'invoices': [
{
'ksefNumber': 'GOOD'
},
{
'ksefNumber': 'BAD'
},
],
}
def get_invoice_by_ksef_number(ksef_number):
return {'xml_content': b'' if ksef_number == 'BAD' else b''}
def mock_parse(self_obj, xml):
if xml == b'':
raise UserError("Simulated error")
return {
'move_type': 'in_invoice',
'invoice_line_ids': [Command.create({'name': 'Test Line', 'price_unit': 100.0})],
}
with (
patch.object(KsefApiService, 'query_invoice_metadata', side_effect=query_invoice_metadata),
patch.object(KsefApiService, 'get_invoice_by_ksef_number', side_effect=get_invoice_by_ksef_number),
patch('odoo.addons.l10n_pl_edi.models.account_move.AccountMove.l10n_pl_edi_get_ksef_bill_vals_from_xml', mock_parse)
):
self.env['account.move'].with_company(self.company)._l10n_pl_edi_download_bills_from_ksef()
bad_invoice = self.env['account.move'].search([('l10n_pl_edi_number', '=', 'BAD')])
good_invoice = self.env['account.move'].search([('l10n_pl_edi_number', '=', 'GOOD')])
self.assertTrue(good_invoice, "Good invoice should be created")
self.assertTrue(bad_invoice, "Bad invoice should be created as empty fallback")
self.assertFalse(bad_invoice.invoice_line_ids, "Bad invoice should have no lines")
self.assertTrue(any("Simulated error" in body for body in bad_invoice.message_ids.mapped('body')))
def test_import_invoice_discount_is_correctly_set(self):
self._assert_import_invoice('invoice_with_discount.xml', [
{
"price_unit": 41.37,
"price_subtotal": 103.43
},
{
"price_unit": 118.3,
"price_subtotal": 1064.7
}
])
def test_l10n_pl_edi_download_bill_same_db(self):
"""
Test that when company_1 sends an invoice to company_2 via KSeF (out_invoice with a ksef number),
company_2 in the same database can still fetch and create the corresponding bill
(in_invoice) with the same KSeF number.
"""
ksef_number = '1234567883-20260123-AAAA1111BBBB-01'
self.standard_invoice.action_post()
self.standard_invoice.l10n_pl_edi_number = ksef_number
def query_invoice_metadata(query_criteria, page_size=100, page_offset=0):
return {
'hasMore': False,
'invoices': [{'ksefNumber': ksef_number}],
}
def get_invoice_by_ksef_number(ksef_nr):
path = 'l10n_pl_edi/tests/export_xmls/fa3_bill.xml'
with tools.file_open(path, mode='rb') as file:
return {'xml_content': file.read()}
# Disable access token for all companies except company_2 to be sure that the cron will try to fetch bills for company_2 only
self.env['res.company'].search([('id', '!=', self.company_2.id)]).l10n_pl_edi_access_token = False
with (
patch.object(KsefApiService, 'query_invoice_metadata', side_effect=query_invoice_metadata),
patch.object(KsefApiService, 'get_invoice_by_ksef_number', side_effect=get_invoice_by_ksef_number) as capt,
):
self.env['account.move'].sudo()._cron_l10n_pl_edi_download_bills()
capt.assert_called_once_with(ksef_number)
bill = self.env['account.move'].search([
('l10n_pl_edi_number', '=', ksef_number),
('company_id', '=', self.company_2.id),
])
self.assertEqual(bill.move_type, 'in_invoice')
moves_count = self.env['account.move'].search_count([('l10n_pl_edi_number', '=', ksef_number)])
self.assertEqual(moves_count, 2)
@freeze_time('2026-01-23')
def test_l10n_pl_edi_download_bill_far_in_the_past(self):
"""
Test that when the last processed bill is far in the past (>3 months),
the download logic slides the 2-month query window forward until it
reaches the current date, rather than issuing a single >3-month query
(which would cause a 400 error from the KSeF API).
"""
old_bill = self.env['account.move'].create({
'move_type': 'in_invoice',
'partner_id': self.partner_pl.id,
'invoice_date': '2025-08-23',
'invoice_line_ids': [Command.create({'product_id': self.product_a.id, 'price_unit': 100.0, 'tax_ids': []})],
})
old_bill.l10n_pl_edi_number = 'OLD-BILL-NUMBER-001'
def query_invoice_metadata(query_criteria, page_size=100, page_offset=0):
date_from = datetime.fromisoformat(query_criteria['dateRange']['from'])
if date_from.month == 12:
return {
'hasMore': False,
'invoices': [{'ksefNumber': 'KSEF-NEW-BILL-001'}],
}
return {'hasMore': False, 'invoices': []}
def get_invoice_by_ksef_number(ksef_number):
path = 'l10n_pl_edi/tests/export_xmls/fa3_bill.xml'
with tools.file_open(path, mode='rb') as file:
return {'xml_content': file.read()}
with (
patch.object(KsefApiService, 'query_invoice_metadata', side_effect=query_invoice_metadata) as capt,
patch.object(KsefApiService, 'get_invoice_by_ksef_number', side_effect=get_invoice_by_ksef_number),
):
self.env['account.move'].with_company(self.company)._l10n_pl_edi_download_bills_from_ksef()
self.assertEqual(capt.call_count, 3)
new_bill = self.env['account.move'].search([('l10n_pl_edi_number', '=', 'KSEF-NEW-BILL-001')])
self.assertTrue(new_bill)
def test_ksef_bill_import_dynamic_taxes_and_fiscal_position(self):
""" Test that importing a KSeF bill maps taxes dynamically using amounts and fiscal positions. """
path = 'l10n_pl_edi/tests/import_xmls/fa3_standard_bill.xml'
with tools.file_open(path, mode='rb') as file:
xml_content = file.read()
parsed_vals = self.env['account.move'].with_company(self.company).l10n_pl_edi_get_ksef_bill_vals_from_xml(xml_content)
bill = self.env['account.move'].with_company(self.company).create(parsed_vals)
self.assertRecordValues(bill.invoice_line_ids, [
{'price_unit': 3.19},
{'price_unit': 5.00},
{'price_unit': 5.00},
])
self.assertEqual(
bill.invoice_line_ids.mapped('tax_ids.amount'),
[23.0, 23.0, 5.0],
)