From 150951f61bded040d671b99e12bbe6dd6e39685f Mon Sep 17 00:00:00 2001 From: Laurent Smet Date: Fri, 19 Aug 2022 08:48:02 +0000 Subject: [PATCH] [IMP] account_edi, l10n_*: Make the EDI engine more flexible Add a new method `_get_move_applicability` allowing to trigger the EDI on any journal entry, using the custom functions you want. closes odoo/odoo#101985 X-original-commit: 0e5626ca5126e6fea7fb95b694229948540764d7 Related: odoo/enterprise#32212 Signed-off-by: Laurent Smet Signed-off-by: Julien Van Roy --- .../models/account_edi_document.py | 115 +++---- .../account_edi/models/account_edi_format.py | 118 +------ addons/account_edi/models/account_move.py | 53 +-- addons/account_edi/tests/common.py | 98 ++---- addons/account_edi/tests/test_edi.py | 317 +++++++----------- .../models/account_edi_format.py | 63 ++-- .../models/account_edi_format.py | 30 +- .../models/account_edi_format.py | 42 +-- .../l10n_in_edi/models/account_edi_format.py | 189 +++++------ .../l10n_it_edi/models/account_edi_format.py | 47 +-- addons/l10n_it_edi/models/account_invoice.py | 2 +- addons/l10n_it_edi/tests/test_edi_export.py | 16 +- .../tests/test_edi_reverse_charge.py | 2 +- 13 files changed, 415 insertions(+), 677 deletions(-) diff --git a/addons/account_edi/models/account_edi_document.py b/addons/account_edi/models/account_edi_document.py index d878907fc3d..a06fd937c4b 100644 --- a/addons/account_edi/models/account_edi_document.py +++ b/addons/account_edi/models/account_edi_document.py @@ -52,10 +52,10 @@ class AccountEdiDocument(models.Model): config_errors = doc.edi_format_id._check_move_configuration(move) if config_errors: res = base64.b64encode('\n'.join(config_errors).encode('UTF-8')) - elif move.is_invoice(include_receipts=True) and doc.edi_format_id._is_required_for_invoice(move): - res = base64.b64encode(doc.edi_format_id._get_invoice_edi_content(doc.move_id)) - elif move.payment_id and doc.edi_format_id._is_required_for_payment(move): - res = base64.b64encode(doc.edi_format_id._get_payment_edi_content(doc.move_id)) + else: + move_applicability = doc.edi_format_id._get_move_applicability(move) + if move_applicability and move_applicability.get('edi_content'): + res = base64.b64encode(move_applicability['edi_content'](move)) doc.edi_content = res def action_export_xml(self): @@ -71,52 +71,44 @@ class AccountEdiDocument(models.Model): doc_type (invoice or payment) and company_id AND the edi_format_id supports batching, they are grouped into a single job. - :returns: A list of tuples (documents, doc_type) - * documents: The documents related to this job. If edi_format_id does not support batch, length is one - * doc_type: Are the moves of this job invoice or payments ? + :returns: [{ + 'documents': account.edi.document, + 'method_to_call': str, + }] """ - # Classify jobs by (edi_format, edi_doc.state, doc_type, move.company_id, custom_key) to_process = {} - documents = self.filtered(lambda d: d.state in ('to_send', 'to_cancel') and d.blocking_level != 'error') - for edi_doc in documents: - move = edi_doc.move_id - edi_format = edi_doc.edi_format_id - if move.is_invoice(include_receipts=True): - doc_type = 'invoice' - elif move.payment_id or move.statement_line_id: - doc_type = 'payment' - else: - continue + for state, edi_flow in (('to_send', 'post'), ('to_cancel', 'cancel')): + documents = self.filtered(lambda d: d.state == state and d.blocking_level != 'error') + for edi_doc in documents: + edi_format = edi_doc.edi_format_id + move = edi_doc.move_id + move_applicability = edi_doc.edi_format_id._get_move_applicability(move) or {} - custom_key = edi_format._get_batch_key(edi_doc.move_id, edi_doc.state) - key = (edi_format, edi_doc.state, doc_type, move.company_id, custom_key) - to_process.setdefault(key, self.env['account.edi.document']) - to_process[key] |= edi_doc - - # Order payments/invoice and create batches. - invoices = [] - payments = [] - for key, documents in to_process.items(): - edi_format, state, doc_type, company_id, custom_key = key - target = invoices if doc_type == 'invoice' else payments - batch = self.env['account.edi.document'] - for doc in documents: - if edi_format._support_batching(move=doc.move_id, state=state, company=company_id): - batch |= doc + batching_key = [edi_format, state, move.company_id] + custom_batching_key = f'{edi_flow}_batching' + if move_applicability.get(custom_batching_key): + batching_key += list(move_applicability[custom_batching_key](move)) else: - target.append((doc, doc_type)) - if batch: - target.append((batch, doc_type)) - return invoices + payments + batching_key.append(move.id) + + batch = to_process.setdefault(tuple(batching_key), { + 'documents': self.env['account.edi.document'], + 'method_to_call': move_applicability.get(edi_flow), + }) + batch['documents'] |= edi_doc + + return list(to_process.values()) @api.model - def _process_job(self, documents, doc_type): + def _process_job(self, job): """Post or cancel move_id (invoice or payment) by calling the related methods on edi_format_id. Invoices are processed before payments. - :param documents: The documents related to this job. If edi_format_id does not support batch, length is one - :param doc_type: Are the moves of this job invoice or payments ? + :param job: { + 'documents': account.edi.document, + 'method_to_call': str, + } """ def _postprocess_post_edi_results(documents, edi_result): attachments_to_unlink = self.env['ir.attachment'] @@ -182,38 +174,36 @@ class AccountEdiDocument(models.Model): # supposed to have any traceability from the user. attachments_to_unlink.sudo().unlink() + documents = job['documents'] + if job['method_to_call']: + method_to_call = job['method_to_call'] + else: + method_to_call = lambda moves: {move: {'success': True} for move in moves} documents.edi_format_id.ensure_one() # All account.edi.document of a job should have the same edi_format_id documents.move_id.company_id.ensure_one() # All account.edi.document of a job should be from the same company if len(set(doc.state for doc in documents)) != 1: raise ValueError('All account.edi.document of a job should have the same state') - edi_format = documents.edi_format_id state = documents[0].state documents.move_id.line_ids.flush_recordset() # manual flush for tax details - if doc_type == 'invoice': - if state == 'to_send': - invoices = documents.move_id - with invoices._send_only_when_ready(): - edi_result = edi_format._post_invoice_edi(invoices) - _postprocess_post_edi_results(documents, edi_result) - elif state == 'to_cancel': - edi_result = edi_format._cancel_invoice_edi(documents.move_id) - _postprocess_cancel_edi_results(documents, edi_result) - - elif doc_type == 'payment': - if state == 'to_send': - edi_result = edi_format._post_payment_edi(documents.move_id) - _postprocess_post_edi_results(documents, edi_result) - elif state == 'to_cancel': - edi_result = edi_format._cancel_payment_edi(documents.move_id) - _postprocess_cancel_edi_results(documents, edi_result) + moves = documents.move_id + if state == 'to_send': + if all(move.is_invoice(include_receipts=True) for move in moves): + with moves._send_only_when_ready(): + edi_result = method_to_call(moves) + else: + edi_result = method_to_call(moves) + _postprocess_post_edi_results(documents, edi_result) + elif state == 'to_cancel': + edi_result = method_to_call(moves) + _postprocess_cancel_edi_results(documents, edi_result) def _process_documents_no_web_services(self): """ Post and cancel all the documents that don't need a web service. """ jobs = self.filtered(lambda d: not d.edi_format_id._needs_web_services())._prepare_jobs() - for documents, doc_type in jobs: - self._process_job(documents, doc_type) + for job in jobs: + self._process_job(job) def _process_documents_web_services(self, job_count=None, with_commit=True): ''' Post and cancel all the documents that need a web service. @@ -225,7 +215,8 @@ class AccountEdiDocument(models.Model): all_jobs = self.filtered(lambda d: d.edi_format_id._needs_web_services())._prepare_jobs() jobs_to_process = all_jobs[0:job_count] if job_count else all_jobs - for documents, doc_type in jobs_to_process: + for job in jobs_to_process: + documents = job['documents'] move_to_lock = documents.move_id attachments_potential_unlink = documents.attachment_id.filtered(lambda a: not a.res_model and not a.res_id) try: @@ -245,7 +236,7 @@ class AccountEdiDocument(models.Model): continue else: raise e - self._process_job(documents, doc_type) + self._process_job(job) if with_commit and len(jobs_to_process) > 1: self.env.cr.commit() diff --git a/addons/account_edi/models/account_edi_format.py b/addons/account_edi/models/account_edi_format.py index 93117c22cfd..ff7613ea955 100644 --- a/addons/account_edi/models/account_edi_format.py +++ b/addons/account_edi/models/account_edi_format.py @@ -28,7 +28,6 @@ class AccountEdiFormat(models.Model): ('unique_code', 'unique (code)', 'This code already exists') ] - #################################################### # Low-level methods #################################################### @@ -51,37 +50,18 @@ class AccountEdiFormat(models.Model): # Export method to override based on EDI Format #################################################### - def _get_invoice_edi_content(self, move): - ''' Create a bytes literal of the file content representing the invoice - to be overridden by the EDI Format - :returns: bytes literal of the content generated (typically XML). - ''' - return b'' + def _get_move_applicability(self, move): + """ Core function for the EDI processing: it first checks whether the EDI format is applicable on a given + move, if so, it then returns a dictionary containing the functions to call for this move. - def _get_payment_edi_content(self, move): - ''' Create a bytes literal of the file content representing the payment - to be overridden by the EDI Format - :returns: bytes literal of the content generated (typically XML). - ''' - return b'' - - def _is_required_for_invoice(self, invoice): - """ Indicate if this EDI must be generated for the invoice passed as parameter. - - :param invoice: An account.move having the invoice type. - :returns: True if the EDI must be generated, False otherwise. + :return: dict mapping str to function (callable) + * post: function called for edi.documents with state 'to_send' (post flow) + * cancel: function called for edi.documents with state 'to_cancel' (cancel flow) + * post_batching: function returning the batching key for the post flow + * cancel_batching: function returning the batching key for the cancel flow + * edi_content: function called when computing the edi_content for an edi.document """ - # TO OVERRIDE self.ensure_one() - return True - - def _is_required_for_payment(self, payment): - """ Indicate if this EDI must be generated for the payment passed as parameter. - - :param payment: An account.move linked to either an account.payment, either an account.bank.statement.line. - :returns: True if the EDI must be generated, False otherwise. - """ - # TO OVERRIDE - self.ensure_one() - return False def _needs_web_services(self): """ Indicate if the EDI must be generated asynchronously through to some web services. @@ -111,33 +91,6 @@ class AccountEdiFormat(models.Model): """ return True - def _support_batching(self, move, state, company): - """ Indicate if we can send multiple documents in the same time to the web services. - If True, the _post_%s_edi methods will get multiple documents in the same time. - Otherwise, these methods will be called with only one record at a time. - - :param move: The move that we are trying to batch. - :param state: The EDI state of the move. - :param company: The company with which we are sending the EDI. - :returns: True if batching is supported, False otherwise. - """ - # TO OVERRIDE - return False - - def _get_batch_key(self, move, state): - """ Returns a tuple that will be used as key to partitionnate the invoices/payments when creating batches - with multiple invoices/payments. - The type of move (invoice or payment), its company_id, its edi state and the edi_format are used by default, if - no further partition is needed for this format, this method should return (). It's not necessary to repeat those - fields in the custom key. - - :param move: The move to batch. - :param state: The EDI state of the move. - :returns: The key to be used when partitionning the batches. - """ - move.ensure_one() - return () - def _check_move_configuration(self, move): """ Checks the move and relevant records for potential error (missing data, etc). @@ -147,59 +100,6 @@ class AccountEdiFormat(models.Model): # TO OVERRIDE return [] - def _post_invoice_edi(self, invoices): - """ Create the file content representing the invoice (and calls web services if necessary). - - :param invoices: A list of invoices to post. - :returns: A dictionary with the invoice as key and as value, another dictionary: - * success: True if the edi was successfully posted. - * attachment: The attachment representing the invoice in this edi_format. - * error: An error if the edi was not successfully posted. - * blocking_level: (optional) How bad is the error (how should the edi flow be blocked ?) - """ - # TO OVERRIDE - self.ensure_one() - return {} - - def _cancel_invoice_edi(self, invoices): - """Calls the web services to cancel the invoice of this document. - - :param invoices: A list of invoices to cancel. - :returns: A dictionary with the invoice as key and as value, another dictionary: - * success: True if the invoice was successfully cancelled. - * error: An error if the edi was not successfully cancelled. - * blocking_level: (optional) How bad is the error (how should the edi flow be blocked ?) - """ - # TO OVERRIDE - self.ensure_one() - return {invoice: {'success': True} for invoice in invoices} # By default, cancel succeeds doing nothing. - - def _post_payment_edi(self, payments): - """ Create the file content representing the payment (and calls web services if necessary). - - :param payments: The payments to post. - :returns: A dictionary with the payment as key and as value, another dictionary: - * attachment: The attachment representing the payment in this edi_format if the edi was successfully posted. - * error: An error if the edi was not successfully posted. - * blocking_level: (optional) How bad is the error (how should the edi flow be blocked ?) - """ - # TO OVERRIDE - self.ensure_one() - return {} - - def _cancel_payment_edi(self, payments): - """Calls the web services to cancel the payment of this document. - - :param payments: A list of payments to cancel. - :returns: A dictionary with the payment as key and as value, another dictionary: - * success: True if the payment was successfully cancelled. - * error: An error if the edi was not successfully cancelled. - * blocking_level: (optional) How bad is the error (how should the edi flow be blocked ?) - """ - # TO OVERRIDE - self.ensure_one() - return {payment: {'success': True} for payment in payments} # By default, cancel succeeds doing nothing. - #################################################### # Import methods to override based on EDI Format #################################################### diff --git a/addons/account_edi/models/account_move.py b/addons/account_edi/models/account_move.py index 81fa9d8ff43..774ea6f2807 100644 --- a/addons/account_edi/models/account_move.py +++ b/addons/account_edi/models/account_move.py @@ -93,10 +93,11 @@ class AccountMove(models.Model): for move in self: for doc in move.edi_document_ids: + move_applicability = doc.edi_format_id._get_move_applicability(move) if doc.edi_format_id._needs_web_services() \ - and doc.state in ('sent', 'to_cancel') \ - and move.is_invoice(include_receipts=True) \ - and doc.edi_format_id._is_required_for_invoice(move): + and doc.state in ('sent', 'to_cancel') \ + and move_applicability \ + and move_applicability.get('cancel'): move.show_reset_to_draft_button = False break @@ -107,20 +108,28 @@ class AccountMove(models.Model): move.edi_show_cancel_button = False continue - move.edi_show_cancel_button = any([doc.edi_format_id._needs_web_services() - and doc.state == 'sent' - and move.is_invoice(include_receipts=True) - and doc.edi_format_id._is_required_for_invoice(move) - for doc in move.edi_document_ids]) + move.edi_show_cancel_button = False + for doc in move.edi_document_ids: + move_applicability = doc.edi_format_id._get_move_applicability(move) + if doc.edi_format_id._needs_web_services() \ + and doc.state == 'sent' \ + and move_applicability \ + and move_applicability.get('cancel'): + move.edi_show_cancel_button = True + break @api.depends('edi_document_ids.state') def _compute_edi_show_abandon_cancel_button(self): for move in self: - move.edi_show_abandon_cancel_button = any(doc.edi_format_id._needs_web_services() - and doc.state == 'to_cancel' - and move.is_invoice(include_receipts=True) - and doc.edi_format_id._is_required_for_invoice(move) - for doc in move.edi_document_ids) + move.edi_show_abandon_cancel_button = False + for doc in move.edi_document_ids: + move_applicability = doc.edi_format_id._get_move_applicability(move) + if doc.edi_format_id._needs_web_services() \ + and doc.state == 'to_cancel' \ + and move_applicability \ + and move_applicability.get('cancel'): + move.edi_show_abandon_cancel_button = True + break #################################################### # Export Electronic Document @@ -250,8 +259,9 @@ class AccountMove(models.Model): edi_formats = self.env['account.edi.format'].browse(edi_formats.ids) # Avoid duplicates for edi_format in edi_formats: existing_edi_document = payment.edi_document_ids.filtered(lambda x: x.edi_format_id == edi_format) + move_applicability = edi_format._get_move_applicability(payment) - if edi_format._is_required_for_payment(payment): + if move_applicability: if existing_edi_document: existing_edi_document.write({ 'state': 'to_send', @@ -293,9 +303,9 @@ class AccountMove(models.Model): edi_document_vals_list = [] for move in posted: for edi_format in move.journal_id.edi_format_ids: - is_edi_needed = move.is_invoice(include_receipts=False) and edi_format._is_required_for_invoice(move) + move_applicability = edi_format._get_move_applicability(move) - if is_edi_needed: + if move_applicability: errors = edi_format._check_move_configuration(move) if errors: raise UserError(_("Invalid invoice configuration:\n\n%s") % '\n'.join(errors)) @@ -352,11 +362,11 @@ class AccountMove(models.Model): for move in self: is_move_marked = False for doc in move.edi_document_ids: + move_applicability = doc.edi_format_id._get_move_applicability(move) if doc.edi_format_id._needs_web_services() \ - and doc.attachment_id \ and doc.state == 'sent' \ - and move.is_invoice(include_receipts=True) \ - and doc.edi_format_id._is_required_for_invoice(move): + and move_applicability \ + and move_applicability.get('cancel'): to_cancel_documents |= doc is_move_marked = True if is_move_marked: @@ -371,9 +381,8 @@ class AccountMove(models.Model): for move in self: is_move_marked = False for doc in move.edi_document_ids: - if doc.state == 'to_cancel' \ - and move.is_invoice(include_receipts=True) \ - and doc.edi_format_id._is_required_for_invoice(move): + move_applicability = doc.edi_format_id._get_move_applicability(move) + if doc.state == 'to_cancel' and move_applicability and move_applicability.get('cancel'): documents |= doc is_move_marked = True if is_move_marked: diff --git a/addons/account_edi/tests/common.py b/addons/account_edi/tests/common.py index 14f08848ca8..f76b884c503 100644 --- a/addons/account_edi/tests/common.py +++ b/addons/account_edi/tests/common.py @@ -5,7 +5,6 @@ from odoo.addons.account.tests.common import AccountTestInvoicingCommon from contextlib import contextmanager from unittest.mock import patch -from unittest import mock import base64 @@ -14,12 +13,17 @@ def _generate_mocked_needs_web_services(needs_web_services): return lambda edi_format: needs_web_services -def _generate_mocked_support_batching(support_batching): - return lambda edi_format, move, state, company: support_batching - - -def _mocked_get_batch_key(edi_format, move, state): - return () +def _mocked_get_move_applicability(edi_format, move): + if move.is_invoice(): + return { + 'post': edi_format._post_invoice_edi, + 'cancel': edi_format._cancel_invoice_edi, + } + elif move.payment_id or move.statement_line_id: + return { + 'post': edi_format._post_payment_edi, + 'cancel': edi_format._cancel_invoice_edi, + } def _mocked_check_move_configuration_success(edi_format, move): @@ -30,48 +34,10 @@ def _mocked_check_move_configuration_fail(edi_format, move): return ['Fake error (mocked)'] -def _mocked_post(edi_format, invoices): - res = {} - for invoice in invoices: - attachment = edi_format.env['ir.attachment'].create({ - 'name': 'mock_simple.xml', - 'datas': base64.encodebytes(b""), - 'mimetype': 'application/xml' - }) - res[invoice] = {'success': True, 'attachment': attachment} - return res - - -def _mocked_post_two_steps(edi_format, invoices): - # For this test, we use the field ref to know if the first step is already done or not. - # Typically, a technical field for the reference of the upload to the web-service will - # be saved on the invoice. - invoices_no_ref = invoices.filtered(lambda i: not i.ref) - if len(invoices_no_ref) == len(invoices): # first step - invoices_no_ref.ref = 'test_ref' - return {invoice: {} for invoice in invoices} - elif len(invoices_no_ref) == 0: # second step - res = {} - for invoice in invoices: - attachment = edi_format.env['ir.attachment'].create({ - 'name': 'mock_simple.xml', - 'datas': base64.encodebytes(b""), - 'mimetype': 'application/xml' - }) - res[invoice] = {'success': True, 'attachment': attachment} - return res - else: - raise ValueError('wrong use of "_mocked_post_two_steps"') - - def _mocked_cancel_success(edi_format, invoices): return {invoice: {'success': True} for invoice in invoices} -def _mocked_cancel_failed(edi_format, invoices): - return {invoice: {'error': 'Faked error (mocked)'} for invoice in invoices} - - class AccountEdiTestCommon(AccountTestInvoicingCommon): @classmethod @@ -93,41 +59,33 @@ class AccountEdiTestCommon(AccountTestInvoicingCommon): # EDI helpers #################################################### + def _create_fake_edi_attachment(self): + return self.env['ir.attachment'].create({ + 'name': '_create_fake_edi_attachment.xml', + 'datas': base64.encodebytes(b""), + 'mimetype': 'application/xml' + }) + + @contextmanager + def with_custom_method(self, method_name, method_content): + path = f'odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat.{method_name}' + with patch(path, new=method_content, create=not hasattr(self.env['account.edi.format'], method_name)): + yield + @contextmanager def mock_edi(self, - _is_required_for_invoice_method=lambda edi_format, invoice: True, - _is_required_for_payment_method=lambda edi_format, invoice: True, - _support_batching_method=_generate_mocked_support_batching(False), - _get_batch_key_method=_mocked_get_batch_key, + _get_move_applicability_method=_mocked_get_move_applicability, _needs_web_services_method=_generate_mocked_needs_web_services(False), _check_move_configuration_method=_mocked_check_move_configuration_success, - _post_invoice_edi_method=_mocked_post, - _cancel_invoice_edi_method=_mocked_cancel_success, - _post_payment_edi_method=_mocked_post, - _cancel_payment_edi_method=_mocked_cancel_success, ): try: - with patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._is_required_for_invoice', - new=_is_required_for_invoice_method), \ - patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._is_required_for_payment', - new=_is_required_for_payment_method), \ - patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._needs_web_services', + with patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._needs_web_services', new=_needs_web_services_method), \ - patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._support_batching', - new=_support_batching_method), \ - patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._get_batch_key', - new=_get_batch_key_method), \ patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._check_move_configuration', new=_check_move_configuration_method), \ - patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._post_invoice_edi', - new=_post_invoice_edi_method), \ - patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._cancel_invoice_edi', - new=_cancel_invoice_edi_method), \ - patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._post_payment_edi', - new=_post_payment_edi_method), \ - patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._cancel_payment_edi', - new=_cancel_payment_edi_method): + patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._get_move_applicability', + new=_get_move_applicability_method): yield finally: diff --git a/addons/account_edi/tests/test_edi.py b/addons/account_edi/tests/test_edi.py index b48a927adf5..d7441e45b27 100644 --- a/addons/account_edi/tests/test_edi.py +++ b/addons/account_edi/tests/test_edi.py @@ -1,9 +1,7 @@ # -*- coding: utf-8 -*- # Part of Odoo. See LICENSE file for full copyright and licensing details. -from odoo import Command -from odoo.addons.account_edi.tests.common import AccountEdiTestCommon, _mocked_post_two_steps, _generate_mocked_needs_web_services, _mocked_cancel_failed, _generate_mocked_support_batching -from unittest.mock import patch +from odoo.addons.account_edi.tests.common import AccountEdiTestCommon from odoo.addons.base.tests.test_ir_cron import CronMixinCase from odoo.tests import tagged @@ -15,197 +13,130 @@ class TestAccountEdi(AccountEdiTestCommon, CronMixinCase): def setUpClass(cls, chart_template_ref=None, edi_format_ref=None): super().setUpClass(chart_template_ref=chart_template_ref, edi_format_ref=edi_format_ref) - cls.invoice = cls.init_invoice('out_invoice', products=cls.product_a + cls.product_b) + cls.env['account.edi.document'].search([]).unlink() + cls.env['account.edi.format'].search([]).unlink() + + cls.test_edi_format = cls.env['account.edi.format'].sudo().create({ + 'name': 'test_edi_format', + 'code': 'test_edi_format', + }) + cls.company_data['default_journal_sale'].edi_format_ids |= cls.test_edi_format def test_export_edi(self): - self.assertEqual(len(self.invoice.edi_document_ids), 0) - self.invoice.action_post() - self.assertEqual(len(self.invoice.edi_document_ids), 1) + invoice = self.init_invoice('out_invoice', products=self.product_a) - def test_prepare_jobs(self): + self.assertEqual(len(invoice.edi_document_ids), 0) + with self.with_custom_method('_get_move_applicability', lambda edi_format, inv: {'post': edi_format._test_edi_post_invoice}), \ + self.with_custom_method('_test_edi_post_invoice', lambda edi_format, inv: {inv: {'success': True}}): + invoice.action_post() + self.assertEqual(len(invoice.edi_document_ids), 1) - edi_docs = self.env['account.edi.document'] - edi_docs |= self.create_edi_document(self.edi_format, 'to_send') - edi_docs |= self.create_edi_document(self.edi_format, 'to_send') + def test_prepare_jobs_no_batching(self): + invoice1 = self.init_invoice('out_invoice', products=self.product_a) + invoice2 = self.init_invoice('out_invoice', products=self.product_a) - to_process = edi_docs._prepare_jobs() - self.assertEqual(len(to_process), 2) + with self.with_custom_method('_get_move_applicability', lambda edi_format, inv: {'post': edi_format._test_edi_post_invoice}), \ + self.with_custom_method('_test_edi_post_invoice', lambda edi_format, inv: {inv: {'success': True}}), \ + self.with_custom_method('_needs_web_services', lambda edi_format: True): + (invoice1 + invoice2).action_post() - with patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._support_batching', return_value=True): - to_process = edi_docs._prepare_jobs() - self.assertEqual(len(to_process), 1) + jobs = (invoice1 + invoice2).edi_document_ids._prepare_jobs() + self.assertEqual(len(jobs), 2) - other_edi = self.env['account.edi.format'].sudo().create({ - 'name': 'Batchable EDI format 2', - 'code': 'test_batch_edi_2', - }) + def test_prepare_jobs_batching(self): + invoice1 = self.init_invoice('out_invoice', products=self.product_a) + invoice2 = self.init_invoice('out_invoice', products=self.product_a) - edi_docs |= self.create_edi_document(other_edi, 'to_send') - edi_docs |= self.create_edi_document(other_edi, 'to_send') + with self.with_custom_method('_get_move_applicability', + lambda edi_format, inv: { + 'post': edi_format._test_edi_post_invoice, + 'post_batching': lambda inv: (inv.partner_id,), + }), \ + self.with_custom_method('_test_edi_post_invoice', lambda edi_format, inv: {inv: {'success': True}}), \ + self.with_custom_method('_needs_web_services', lambda edi_format: True): + (invoice1 + invoice2).action_post() - with patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._support_batching', return_value=True): - to_process = edi_docs._prepare_jobs() - self.assertEqual(len(to_process), 2) + jobs = (invoice1 + invoice2).edi_document_ids._prepare_jobs() + self.assertEqual(len(jobs), 1) - @patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._post_invoice_edi', return_value={}) - def test_warning_is_retried(self, patched): - with patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._needs_web_services', - new=lambda edi_format: True): - edi_docs = self.create_edi_document(self.edi_format, 'to_send') - edi_docs.error = 'Test Error' - edi_docs.blocking_level = 'warning' + def test_warning_is_retried(self): + invoice = self.init_invoice('out_invoice', products=self.product_a) - edi_docs.move_id.action_process_edi_web_services() - patched.assert_called_once() + with self.with_custom_method('_get_move_applicability', lambda edi_format, inv: {'post': edi_format._test_edi_post_invoice}), \ + self.with_custom_method('_needs_web_services', lambda edi_format: True): + + with self.with_custom_method('_test_edi_post_invoice', + lambda edi_format, inv: {inv: {'error': "turlututu", 'blocking_level': 'warning'}}): + invoice.action_post() + self.assertRecordValues(invoice.edi_document_ids, [{'state': 'to_send'}]) + + with self.with_custom_method('_test_edi_post_invoice', lambda edi_format, inv: {inv: {'success': True}}): + invoice.action_process_edi_web_services(with_commit=False) + self.assertRecordValues(invoice.edi_document_ids, [{'state': 'sent'}]) def test_edi_flow(self): - with self.mock_edi(): - doc = self.invoice._get_edi_document(self.edi_format) - self.assertFalse(doc) - self.invoice.action_post() - doc = self.invoice._get_edi_document(self.edi_format) - self.assertEqual(len(doc), 1) - self.assertEqual(doc.state, 'sent') - self.invoice.button_draft() - self.invoice.button_cancel() - self.assertEqual(doc.state, 'cancelled') + invoice = self.init_invoice('out_invoice', products=self.product_a) + + with self.with_custom_method('_get_move_applicability', lambda edi_format, inv: { + 'post': edi_format._test_edi_post_invoice, + 'cancel': edi_format._test_edi_cancel_invoice, + }), \ + self.with_custom_method('_needs_web_services', lambda edi_format: True), \ + self.with_custom_method('_test_edi_post_invoice', lambda edi_format, inv: {inv: {'success': True}}), \ + self.with_custom_method('_test_edi_cancel_invoice', lambda edi_format, inv: {inv: {'success': True}}): + invoice.action_post() + self.assertRecordValues(invoice.edi_document_ids, [{'state': 'to_send'}]) + + invoice.action_process_edi_web_services(with_commit=False) + self.assertRecordValues(invoice.edi_document_ids, [{'state': 'sent'}]) + + invoice.button_cancel_posted_moves() + self.assertRecordValues(invoice.edi_document_ids, [{'state': 'to_cancel'}]) + + invoice.button_abandon_cancel_posted_posted_moves() + self.assertRecordValues(invoice.edi_document_ids, [{'state': 'sent'}]) + + invoice.button_cancel_posted_moves() + self.assertRecordValues(invoice.edi_document_ids, [{'state': 'to_cancel'}]) + + invoice.action_process_edi_web_services(with_commit=False) + self.assertRecordValues(invoice.edi_document_ids, [{'state': 'cancelled'}]) def test_edi_flow_two_steps(self): - with self.mock_edi(_post_invoice_edi_method=_mocked_post_two_steps, - _needs_web_services_method=_generate_mocked_needs_web_services(True)): - doc = self.invoice._get_edi_document(self.edi_format) - self.assertFalse(doc) - self.invoice.action_post() - doc = self.invoice._get_edi_document(self.edi_format) - self.assertEqual(len(doc), 1) - self.assertEqual(doc.state, 'to_send') - doc._process_documents_web_services(with_commit=False) - self.assertEqual(doc.state, 'to_send') - doc._process_documents_web_services(with_commit=False) - self.assertEqual(doc.state, 'sent') + def step1(edi_format, invoice): + return {invoice: {'error': "step1 done", 'blocking_level': 'info'}} - def test_edi_flow_request_cancel_success(self): - with self.mock_edi(_needs_web_services_method=_generate_mocked_needs_web_services(True)): - self.assertEqual(self.invoice.state, 'draft') - self.invoice.action_post() - doc = self.invoice._get_edi_document(self.edi_format) - self.assertEqual(doc.state, 'to_send') - self.assertEqual(self.invoice.state, 'posted') - doc._process_documents_web_services(with_commit=False) - self.assertEqual(doc.state, 'sent') - self.assertEqual(self.invoice.state, 'posted') - self.invoice.button_cancel_posted_moves() - self.assertEqual(doc.state, 'to_cancel') - self.assertEqual(self.invoice.state, 'posted') - doc._process_documents_web_services(with_commit=False) - self.assertEqual(doc.state, 'cancelled') - self.assertEqual(self.invoice.state, 'cancel') + def step2(edi_format, invoice): + return {invoice: {'success': True}} - def test_edi_flow_request_cancel_failed(self): - with self.mock_edi(_needs_web_services_method=_generate_mocked_needs_web_services(True), - _cancel_invoice_edi_method=_mocked_cancel_failed): - self.assertEqual(self.invoice.state, 'draft') - self.invoice.action_post() - doc = self.invoice._get_edi_document(self.edi_format) - self.assertEqual(doc.state, 'to_send') - self.assertEqual(self.invoice.state, 'posted') - doc._process_documents_web_services(with_commit=False) - self.assertEqual(doc.state, 'sent') - self.assertEqual(self.invoice.state, 'posted') - self.invoice.button_cancel_posted_moves() - self.assertEqual(doc.state, 'to_cancel') - self.assertEqual(self.invoice.state, 'posted') - # Call off edi Cancellation - self.invoice.button_abandon_cancel_posted_posted_moves() - self.assertEqual(doc.state, 'sent') - self.assertFalse(doc.error) - - # Failed cancel - self.invoice.button_cancel_posted_moves() - self.assertEqual(doc.state, 'to_cancel') - self.assertEqual(self.invoice.state, 'posted') - doc._process_documents_web_services(with_commit=False) - self.assertEqual(doc.state, 'to_cancel') - self.assertEqual(self.invoice.state, 'posted') - - # Call off edi Cancellation - self.invoice.button_abandon_cancel_posted_posted_moves() - self.assertEqual(doc.state, 'sent') - self.assertIsNotNone(doc.error) - - def test_edi_flow_two_step_cancel_with_call_off_request(self): - def _mock_cancel(edi_format, invoices): - invoices_no_ref = invoices.filtered(lambda i: not i.ref) - if len(invoices_no_ref) == len(invoices): # first step - invoices_no_ref.ref = 'test_ref_cancel' - return {invoice: {} for invoice in invoices} - elif len(invoices_no_ref) == 0: # second step - for invoice in invoices: - invoice.ref = None - return {invoice: {'success': True} for invoice in invoices} + def get_move_applicability(edi_format, invoice): + if "step1" in (invoice.edi_document_ids.error or ''): + return {'post': edi_format._test_edi_post_invoice_step2} else: - raise ValueError('wrong use of "_mocked_post_two_steps"') + return {'post': edi_format._test_edi_post_invoice_step1} - def _is_needed_for_invoice(edi_format, invoice): - return not bool(invoice.ref) + invoice = self.init_invoice('out_invoice', products=self.product_a) - with self.mock_edi(_needs_web_services_method=_generate_mocked_needs_web_services(True), - _is_required_for_invoice_method=_is_needed_for_invoice, - _cancel_invoice_edi_method=_mock_cancel): - self.invoice.action_post() - doc = self.invoice._get_edi_document(self.edi_format) - doc._process_documents_web_services(with_commit=False) - self.assertEqual(doc.state, 'sent') + with self.with_custom_method('_get_move_applicability', get_move_applicability), \ + self.with_custom_method('_needs_web_services', lambda edi_format: True), \ + self.with_custom_method('_test_edi_post_invoice_step1', step1), \ + self.with_custom_method('_test_edi_post_invoice_step2', step2): + invoice.action_post() + self.assertRecordValues(invoice.edi_document_ids, [{'state': 'to_send'}]) - # Request Cancellation - self.invoice.button_cancel_posted_moves() - doc._process_documents_web_services(with_commit=False) # first step of cancel - self.assertEqual(doc.state, 'to_cancel') + invoice.action_process_edi_web_services(with_commit=False) + self.assertRecordValues(invoice.edi_document_ids, [{'state': 'to_send'}]) - # Call off edi Cancellation - self.invoice.button_abandon_cancel_posted_posted_moves() - self.assertEqual(doc.state, 'to_cancel') - - # If we cannot call off edi cancellation, only solution is to post again - doc._process_documents_web_services(with_commit=False) # second step of cancel - self.assertEqual(doc.state, 'cancelled') - self.invoice.action_post() - doc._process_documents_web_services(with_commit=False) - self.assertEqual(doc.state, 'sent') - - def test_batches(self): - def _get_batch_key_method(edi_format, move, state): - return (move.ref) - - with self.mock_edi(_get_batch_key_method=_get_batch_key_method, - _support_batching_method=_generate_mocked_support_batching(True)): - edi_docs = self.env['account.edi.document'] - doc1 = self.create_edi_document(self.edi_format, 'to_send') - edi_docs |= doc1 - doc2 = self.create_edi_document(self.edi_format, 'to_send') - edi_docs |= doc2 - doc3 = self.create_edi_document(self.edi_format, 'to_send') - edi_docs |= doc3 - - to_process = edi_docs._prepare_jobs() - self.assertEqual(len(to_process), 1) - - doc1.move_id.ref = 'batch1' - doc2.move_id.ref = 'batch2' - doc3.move_id.ref = 'batch3' - - to_process = edi_docs._prepare_jobs() - self.assertEqual(len(to_process), 3) - - doc2.move_id.ref = 'batch1' - to_process = edi_docs._prepare_jobs() - self.assertEqual(len(to_process), 2) + invoice.action_process_edi_web_services(with_commit=False) + self.assertRecordValues(invoice.edi_document_ids, [{'state': 'sent'}]) def test_cron_triggers(self): - with self.capture_triggers('account_edi.ir_cron_edi_network') as capt, \ - self.mock_edi(_needs_web_services_method=_generate_mocked_needs_web_services(True)): - self.invoice._get_edi_document(self.edi_format) - self.invoice.action_post() + invoice = self.init_invoice('out_invoice', products=self.product_a) + with self.with_custom_method('_get_move_applicability', lambda edi_format, inv: {'post': edi_format._test_edi_post_invoice}), \ + self.with_custom_method('_needs_web_services', lambda edi_format: True), \ + self.with_custom_method('_test_edi_post_invoice', lambda edi_format, inv: {inv: {'success': True}}), \ + self.capture_triggers('account_edi.ir_cron_edi_network') as capt: + invoice.action_post() capt.records.ensure_one() def test_cron_self_trigger(self): @@ -213,33 +144,27 @@ class TestAccountEdi(AccountEdiTestCommon, CronMixinCase): edi_cron = self.env.ref('account_edi.ir_cron_edi_network') edi_cron.code = 'model._cron_process_documents_web_services(job_count=1)' - # Create invoices. - invoices = self.env['account.move'].create([{ - 'move_type': 'out_invoice', - 'invoice_date': '2019-01-01', - 'date': '2019-01-01', - 'partner_id': self.partner_a.id, - 'invoice_line_ids': [Command.create({'product_id': self.product_a.id})], - } for i in range(4)]) - - with self.capture_triggers('account_edi.ir_cron_edi_network') as capt, \ - self.mock_edi(_needs_web_services_method=_generate_mocked_needs_web_services(True)): - invoices.action_post() + invoice1 = self.init_invoice('out_invoice', products=self.product_a) + invoice2 = self.init_invoice('out_invoice', products=self.product_a) + with self.with_custom_method('_get_move_applicability', lambda edi_format, inv: {'post': edi_format._test_edi_post_invoice}), \ + self.with_custom_method('_needs_web_services', lambda edi_format: True), \ + self.with_custom_method('_test_edi_post_invoice', lambda edi_format, inv: {inv: {'success': True}}), \ + self.capture_triggers('account_edi.ir_cron_edi_network') as capt: + (invoice1 + invoice2).action_post() self.env.ref('account_edi.ir_cron_edi_network').method_direct_trigger() - self.assertEqual(len(capt.records), 2, "Not all records have been processed in this run, the cron should " - "re-trigger itself to process some more later") + self.assertEqual( + len(capt.records), 2, + "Not all records have been processed in this run, the cron should re-trigger itself to process some" + " more later", + ) def test_invoice_ready_to_be_sent(self): - def _is_needed_for_invoice(edi_format, invoice): - return True - - with self.mock_edi( - _needs_web_services_method=_generate_mocked_needs_web_services(True), - _is_required_for_invoice_method=_is_needed_for_invoice, - ): - self.invoice.action_post() - doc = self.invoice._get_edi_document(self.edi_format) - self.assertFalse(self.invoice._is_ready_to_be_sent()) - doc._process_documents_web_services(with_commit=False) - self.assertTrue(self.invoice._is_ready_to_be_sent()) + invoice = self.init_invoice('out_invoice', products=self.product_a) + with self.with_custom_method('_get_move_applicability', lambda edi_format, inv: {'post': edi_format._test_edi_post_invoice}), \ + self.with_custom_method('_needs_web_services', lambda edi_format: True), \ + self.with_custom_method('_test_edi_post_invoice', lambda edi_format, inv: {inv: {'success': True}}): + invoice.action_post() + self.assertFalse(invoice._is_ready_to_be_sent()) + invoice.action_process_edi_web_services(with_commit=False) + self.assertTrue(invoice._is_ready_to_be_sent()) diff --git a/addons/account_edi_ubl_cii/models/account_edi_format.py b/addons/account_edi_ubl_cii/models/account_edi_format.py index 920f0d3f765..7bfdae7aee0 100644 --- a/addons/account_edi_ubl_cii/models/account_edi_format.py +++ b/addons/account_edi_ubl_cii/models/account_edi_format.py @@ -74,14 +74,6 @@ class AccountEdiFormat(models.Model): # Export: Account.edi.format override #################################################### - def _is_required_for_invoice(self, invoice): - # EXTENDS account_edi - self.ensure_one() - if self.code not in FORMAT_CODES: - return super()._is_required_for_invoice(invoice) - - return self._is_ubl_cii_available(invoice.company_id) and invoice.move_type in ('out_invoice', 'out_refund') - def _is_compatible_with_journal(self, journal): # EXTENDS account_edi # the formats appear on the journal only if they are compatible (e.g. NLCIUS only appear for dutch companies) @@ -98,39 +90,38 @@ class AccountEdiFormat(models.Model): return super()._is_enabled_by_default_on_journal(journal) return self.code == 'facturx_1_0_05' - def _post_invoice_edi(self, invoices): + def _ubl_cii_post_invoice(self, invoice): # EXTENDS account_edi self.ensure_one() + builder = self._get_xml_builder(invoice.company_id) + # For now, the errors are not displayed anywhere, don't want to annoy the user + xml_content, _ = builder._export_invoice(invoice) + + # DEBUG: send directly to the test platform (the one used by ecosio) + #response = self.env['account.edi.common']._check_xml_ecosio(invoice, xml_content, builder._export_invoice_ecosio_schematrons()) + + attachment_create_vals = { + 'name': builder._export_invoice_filename(invoice), + 'raw': xml_content, + 'mimetype': 'application/xml', + } + # we don't want the Factur-X and E-FFF xml to appear in the attachment of the invoice when confirming it + # E-FFF will appear after the pdf is generated, Factur-X will never appear (it's contained in the PDF) + if self.code not in ['facturx_1_0_05', 'efff_1']: + attachment_create_vals.update({'res_id': invoice.id, 'res_model': 'account.move'}) + + attachment = self.env['ir.attachment'].create(attachment_create_vals) + return {invoice: {'success': True, 'attachment': attachment}} + + def _get_move_applicability(self, move): + # EXTENDS account_edi + self.ensure_one() if self.code not in FORMAT_CODES: - return super()._post_invoice_edi(invoices) + return super()._get_move_applicability(move) - res = {} - for invoice in invoices: - builder = self._get_xml_builder(invoice.company_id) - # For now, the errors are not displayed anywhere, don't want to annoy the user - xml_content, errors = builder._export_invoice(invoice) - - # DEBUG: send directly to the test platform (the one used by ecosio) - #response = self.env['account.edi.common']._check_xml_ecosio(invoice, xml_content, builder._export_invoice_ecosio_schematrons()) - - attachment_create_vals = { - 'name': builder._export_invoice_filename(invoice), - 'raw': xml_content, - 'mimetype': 'application/xml', - } - # we don't want the Factur-X, E-FFF and NLCIUS xml to appear in the attachment of the invoice when confirming it - # E-FFF and NLCIUS will appear after the pdf is generated, Factur-X will never appear (it's contained in the PDF) - if self.code not in ['facturx_1_0_05', 'efff_1', 'nlcius_1']: - attachment_create_vals.update({'res_id': invoice.id, 'res_model': 'account.move'}) - - attachment = self.env['ir.attachment'].create(attachment_create_vals) - res[invoice] = { - 'success': True, - 'attachment': attachment, - } - - return res + if self._is_ubl_cii_available(move.company_id) and move.move_type in ('out_invoice', 'out_refund'): + return {'post': self._ubl_cii_post_invoice} def _is_embedding_to_invoice_pdf_needed(self): # EXTENDS account_edi diff --git a/addons/l10n_eg_edi_eta/models/account_edi_format.py b/addons/l10n_eg_edi_eta/models/account_edi_format.py index a270cdea91a..d2b5c192754 100644 --- a/addons/l10n_eg_edi_eta/models/account_edi_format.py +++ b/addons/l10n_eg_edi_eta/models/account_edi_format.py @@ -311,6 +311,19 @@ class AccountEdiFormat(models.Model): def _needs_web_services(self): return self.code == 'eg_eta' or super()._needs_web_services() + def _get_move_applicability(self, move): + # EXTENDS account_edi + self.ensure_one() + if self.code != 'eg_eta': + return super()._get_move_applicability(move) + + if move.is_invoice(include_receipts=True) and move.country_code == 'EG': + return { + 'post': self._l10n_eg_edi_post_invoice, + 'cancel': self._l10n_eg_edi_cancel_invoice, + 'edi_content': self._l10n_eg_edi_xml_invoice_content, + } + def _check_move_configuration(self, invoice): errors = super()._check_move_configuration(invoice) if self.code != 'eg_eta': @@ -336,11 +349,7 @@ class AccountEdiFormat(models.Model): errors.append(_("Please make sure the EGS/GS1 Barcode is set correctly on all products")) return errors - def _post_invoice_edi(self, invoices): - if self.code != 'eg_eta': - return super()._post_invoice_edi(invoices) - invoice = invoices # Batching is disabled for this EDI. - + def _l10n_eg_edi_post_invoice(self, invoice): # In case we have already sent it, but have not got a final answer yet. if invoice.l10n_eg_submission_number: return {invoice: self._l10n_eg_get_einvoice_status(invoice)} @@ -362,16 +371,11 @@ class AccountEdiFormat(models.Model): } return {invoice: self._l10n_eg_edi_post_invoice_web_service(invoice)} - def _cancel_invoice_edi(self, invoices): - if self.code != 'eg_eta': - return super()._cancel_invoice_edi(invoices) - invoice = invoices + def _l10n_eg_edi_cancel_invoice(self, invoice): return {invoice: self._cancel_invoice_edi_eta(invoice)} - def _get_invoice_edi_content(self, move): - if self.code != 'eg_eta': - return super()._get_invoice_edi_content(move) - return json.dumps(self._l10n_eg_eta_prepare_eta_invoice(move)).encode() + def _l10n_eg_edi_xml_invoice_content(self, invoice): + return json.dumps(self._l10n_eg_eta_prepare_eta_invoice(invoice)).encode() def _is_compatible_with_journal(self, journal): # OVERRIDE diff --git a/addons/l10n_es_edi_sii/models/account_edi_format.py b/addons/l10n_es_edi_sii/models/account_edi_format.py index 2e596d19473..552f63208f1 100644 --- a/addons/l10n_es_edi_sii/models/account_edi_format.py +++ b/addons/l10n_es_edi_sii/models/account_edi_format.py @@ -440,7 +440,7 @@ class AccountEdiFormat(models.Model): def _l10n_es_edi_call_web_service_sign(self, invoices, info_list): company = invoices.company_id - # All are sharing the same value, see '_get_batch_key'. + # All are sharing the same value. csv_number = invoices.mapped('l10n_es_edi_csv')[0] # Set registration date @@ -573,36 +573,26 @@ class AccountEdiFormat(models.Model): # EDI OVERRIDDEN METHODS # ------------------------------------------------------------------------- - def _get_invoice_edi_content(self, move): - if self.code != 'es_sii': - return super()._get_invoice_edi_content(move) - return json.dumps(self._l10n_es_edi_get_invoices_info(move)).encode() + def _l10n_es_edi_sii_xml_invoice_content(self, invoice): + return json.dumps(self._l10n_es_edi_get_invoices_info(invoice)).encode() - def _is_required_for_invoice(self, invoice): - # OVERRIDE + def _get_move_applicability(self, move): + # EXTENDS account_edi + self.ensure_one() if self.code != 'es_sii': - return super()._is_required_for_invoice(invoice) + return super()._get_move_applicability(move) - return invoice.l10n_es_edi_is_required + if move.l10n_es_edi_is_required: + return { + 'post': self._l10n_es_edi_sii_post_invoices, + 'post_batching': lambda invoice: (invoice.move_type, invoice.l10n_es_edi_csv), + 'edi_content': self._l10n_es_edi_sii_xml_invoice_content, + } def _needs_web_services(self): # OVERRIDE return self.code == 'es_sii' or super()._needs_web_services() - def _support_batching(self, move=None, state=None, company=None): - # OVERRIDE - if self.code != 'es_sii': - return super()._support_batching(move=move, state=state, company=company) - - return state == 'to_send' and move.is_invoice() - - def _get_batch_key(self, move, state): - # OVERRIDE - if self.code != 'es_sii': - return super()._get_batch_key(move, state) - - return move.move_type, move.l10n_es_edi_csv - def _check_move_configuration(self, move): # OVERRIDE res = super()._check_move_configuration(move) @@ -642,11 +632,7 @@ class AccountEdiFormat(models.Model): return journal.country_code == 'ES' - def _post_invoice_edi(self, invoices): - # OVERRIDE - if self.code != 'es_sii': - return super()._post_invoice_edi(invoices) - + def _l10n_es_edi_sii_post_invoices(self, invoices): # Ensure a certificate is available. certificate = invoices.company_id.l10n_es_edi_certificate_id if not certificate: diff --git a/addons/l10n_in_edi/models/account_edi_format.py b/addons/l10n_in_edi/models/account_edi_format.py index f239348a557..2412df9301b 100644 --- a/addons/l10n_in_edi/models/account_edi_format.py +++ b/addons/l10n_in_edi/models/account_edi_format.py @@ -27,28 +27,31 @@ class AccountEdiFormat(models.Model): return journal.company_id.country_id.code == 'IN' return super()._is_enabled_by_default_on_journal(journal) - def _is_required_for_invoice(self, invoice): + def _get_move_applicability(self, move): + # EXTENDS account_edi self.ensure_one() - if self.code == "in_einvoice_1_03": - return invoice.is_sale_document() and invoice.country_code == 'IN' and invoice.l10n_in_gst_treatment in ( - "regular", - "composition", - "overseas", - "special_economic_zone", - "deemed_export", - "uin_holders", - ) - return super()._is_required_for_invoice(invoice) + if self.code != 'in_einvoice_1_03': + return super()._get_move_applicability(move) + + if move.is_sale_document() and move.country_code == 'IN' and move.l10n_in_gst_treatment in ( + "regular", + "composition", + "overseas", + "special_economic_zone", + "deemed_export", + ): + return { + 'post': self._l10n_in_edi_post_invoice, + 'cancel': self._l10n_in_edi_cancel_invoice, + 'edi_content': self._l10n_in_edi_xml_invoice_content, + } def _needs_web_services(self): self.ensure_one() return self.code == "in_einvoice_1_03" or super()._needs_web_services() - def _get_invoice_edi_content(self, move): - if self.code != "in_einvoice_1_03": - return super()._get_invoice_edi_content(move) - json_dump = json.dumps(self._l10n_in_edi_generate_invoice_json(move)) - return json_dump.encode() + def _l10n_in_edi_xml_invoice_content(self, invoice): + return json.dumps(self._l10n_in_edi_generate_invoice_json(invoice)).encode() def _l10n_in_edi_extract_digits(self, string): if not string: @@ -88,128 +91,118 @@ class AccountEdiFormat(models.Model): _("Buy Credits") ) - def _post_invoice_edi(self, invoices): - if self.code != "in_einvoice_1_03": - return super()._post_invoice_edi(invoices) - response = {} - res = {} - generate_json = self._l10n_in_edi_generate_invoice_json(invoices) - response = self._l10n_in_edi_generate(invoices.company_id, generate_json) + def _l10n_in_edi_post_invoice(self, invoice): + generate_json = self._l10n_in_edi_generate_invoice_json(invoice) + response = self._l10n_in_edi_generate(invoice.company_id, generate_json) if response.get("error"): error = response["error"] error_codes = [e.get("code") for e in error] if "1005" in error_codes: # Invalid token eror then create new token and send generate request again. # This happen when authenticate called from another odoo instance with same credentials (like. Demo/Test) - authenticate_response = self._l10n_in_edi_authenticate(invoices.company_id) + authenticate_response = self._l10n_in_edi_authenticate(invoice.company_id) if not authenticate_response.get("error"): error = [] - response = self._l10n_in_edi_generate(invoices.company_id, generate_json) + response = self._l10n_in_edi_generate(invoice.company_id, generate_json) if response.get("error"): error = response["error"] error_codes = [e.get("code") for e in error] if "2150" in error_codes: # Get IRN by details in case of IRN is already generated # this happens when timeout from the Government portal but IRN is generated - response = self._l10n_in_edi_get_irn_by_details(invoices.company_id, { - "doc_type": invoices.move_type == "out_refund" and "CRN" or "INV", - "doc_num": invoices.name, - "doc_date": invoices.invoice_date and invoices.invoice_date.strftime("%d/%m/%Y") or False, + response = self._l10n_in_edi_get_irn_by_details(invoice.company_id, { + "doc_type": invoice.move_type == "out_refund" and "CRN" or "INV", + "doc_num": invoice.name, + "doc_date": invoice.invoice_date and invoice.invoice_date.strftime("%d/%m/%Y") or False, }) if not response.get("error"): error = [] odoobot = self.env.ref("base.partner_root") - invoices.message_post(author_id=odoobot.id, body=_( + invoice.message_post(author_id=odoobot.id, body=_( "Somehow this invoice had been submited to government before." \ "
Normally, this should not happen too often" \ "
Just verify value of invoice by uploade json to government website " \ "here." )) if "no-credit" in error_codes: - res[invoices] = { + return {invoice: { "success": False, - "error": self._l10n_in_edi_get_iap_buy_credits_message(invoices.company_id), + "error": self._l10n_in_edi_get_iap_buy_credits_message(invoice.company_id), "blocking_level": "error", - } + }} elif error: error_message = "
".join(["[%s] %s" % (e.get("code"), html_escape(e.get("message"))) for e in error]) - res[invoices] = { + return {invoice: { "success": False, "error": error_message, "blocking_level": ("404" in error_codes) and "warning" or "error", - } + }} if not response.get("error"): json_dump = json.dumps(response.get("data")) - json_name = "%s_einvoice.json" % (invoices.name.replace("/", "_")) + json_name = "%s_einvoice.json" % (invoice.name.replace("/", "_")) attachment = self.env["ir.attachment"].create({ "name": json_name, "raw": json_dump.encode(), "res_model": "account.move", - "res_id": invoices.id, + "res_id": invoice.id, "mimetype": "application/json", }) - res[invoices] = {"success": True, "attachment": attachment} - return res + return {invoice: {"success": True, "attachment": attachment}} - def _cancel_invoice_edi(self, invoices): - if self.code != "in_einvoice_1_03": - return super()._cancel_invoice_edi(invoices) - res = {} - for invoice in invoices: - l10n_in_edi_response_json = invoice._get_l10n_in_edi_response_json() - cancel_json = { - "Irn": l10n_in_edi_response_json.get("Irn"), - "CnlRsn": invoice.l10n_in_edi_cancel_reason, - "CnlRem": invoice.l10n_in_edi_cancel_remarks, - } - response = self._l10n_in_edi_cancel(invoice.company_id, cancel_json) - if response.get("error"): - error = response["error"] - error_codes = [e.get("code") for e in error] - if "1005" in error_codes: - # Invalid token eror then create new token and send generate request again. - # This happen when authenticate called from another odoo instance with same credentials (like. Demo/Test) - authenticate_response = self._l10n_in_edi_authenticate(invoice.company_id) - if not authenticate_response.get("error"): - error = [] - response = self._l10n_in_edi_cancel(invoice.company_id, cancel_json) - if response.get("error"): - error = response["error"] - error_codes = [e.get("code") for e in error] - if "9999" in error_codes: - response = {} - odoobot = self.env.ref("base.partner_root") - invoices.message_post(author_id=odoobot.id, body=_( - "Somehow this invoice had been cancelled to government before." \ - "
Normally, this should not happen too often" \ - "
Just verify by logging into government website " \ - "here." - )) - if "no-credit" in error_codes: - res[invoice] = { - "success": False, - "error": self._l10n_in_edi_get_iap_buy_credits_message(invoice.company_id), - "blocking_level": "error", - } - else: - error_message = "
".join(["[%s] %s" % (e.get("code"), html_escape(e.get("message"))) for e in error]) - res[invoice] = { - "success": False, - "error": error_message, - "blocking_level": ("404" in error_codes) and "warning" or "error", - } - if not response.get("error"): - json_dump = json.dumps(response.get("data", {})) - json_name = "%s_cancel_einvoice.json" % (invoice.name.replace("/", "_")) - attachment = self.env["ir.attachment"].create({ - "name": json_name, - "raw": json_dump.encode(), - "res_model": "account.move", - "res_id": invoice.id, - "mimetype": "application/json", - }) - res[invoice] = {"success": True, "attachment": attachment} - return res + def _l10n_in_edi_cancel_invoice(self, invoice): + l10n_in_edi_response_json = invoice._get_l10n_in_edi_response_json() + cancel_json = { + "Irn": l10n_in_edi_response_json.get("Irn"), + "CnlRsn": invoice.l10n_in_edi_cancel_reason, + "CnlRem": invoice.l10n_in_edi_cancel_remarks, + } + response = self._l10n_in_edi_cancel(invoice.company_id, cancel_json) + if response.get("error"): + error = response["error"] + error_codes = [e.get("code") for e in error] + if "1005" in error_codes: + # Invalid token eror then create new token and send generate request again. + # This happen when authenticate called from another odoo instance with same credentials (like. Demo/Test) + authenticate_response = self._l10n_in_edi_authenticate(invoice.company_id) + if not authenticate_response.get("error"): + error = [] + response = self._l10n_in_edi_cancel(invoice.company_id, cancel_json) + if response.get("error"): + error = response["error"] + error_codes = [e.get("code") for e in error] + if "9999" in error_codes: + response = {} + odoobot = self.env.ref("base.partner_root") + invoice.message_post(author_id=odoobot.id, body=_( + "Somehow this invoice had been cancelled to government before." \ + "
Normally, this should not happen too often" \ + "
Just verify by logging into government website " \ + "here." + )) + if "no-credit" in error_codes: + return {invoice: { + "success": False, + "error": self._l10n_in_edi_get_iap_buy_credits_message(invoice.company_id), + "blocking_level": "error", + }} + else: + error_message = "
".join(["[%s] %s" % (e.get("code"), html_escape(e.get("message"))) for e in error]) + return {invoice: { + "success": False, + "error": error_message, + "blocking_level": ("404" in error_codes) and "warning" or "error", + }} + if not response.get("error"): + json_dump = json.dumps(response.get("data", {})) + json_name = "%s_cancel_einvoice.json" % (invoice.name.replace("/", "_")) + attachment = self.env["ir.attachment"].create({ + "name": json_name, + "raw": json_dump.encode(), + "res_model": "account.move", + "res_id": invoice.id, + "mimetype": "application/json", + }) + return {invoice: {"success": True, "attachment": attachment}} def _l10n_in_validate_partner(self, partner, is_company=False): self.ensure_one() diff --git a/addons/l10n_it_edi/models/account_edi_format.py b/addons/l10n_it_edi/models/account_edi_format.py index df2a1573cd5..acb1f8321fb 100644 --- a/addons/l10n_it_edi/models/account_edi_format.py +++ b/addons/l10n_it_edi/models/account_edi_format.py @@ -790,19 +790,20 @@ class AccountEdiFormat(models.Model): return super()._is_compatible_with_journal(journal) return journal.type in ('sale', 'purchase') and journal.country_code == 'IT' - def _is_required_for_invoice(self, invoice): + def _get_move_applicability(self, move): # OVERRIDE self.ensure_one() if self.code != 'fattura_pa': - return super()._is_required_for_invoice(invoice) + return super()._get_move_applicability(move) - is_self_invoice = self._l10n_it_edi_is_self_invoice(invoice) - return ( - (invoice.is_sale_document() or (is_self_invoice and invoice.is_purchase_document())) - and invoice.country_code == 'IT' - ) + is_it_purchase_document = self._l10n_it_edi_is_self_invoice(move) and move.is_purchase_document() + if move.country_code == 'IT' and (move.is_sale_document() or is_it_purchase_document): + return { + 'post': self._post_fattura_pa, + 'post_batching': lambda move: (move.move_type, bool(move.l10n_it_edi_transaction)), + } - def _export_as_xml(self, invoice): + def _l10n_it_edi_export_invoice_as_xml(self, invoice): ''' Create the xml file content. :return: The XML content as str. ''' @@ -817,12 +818,6 @@ class AccountEdiFormat(models.Model): )) return content - def _get_invoice_edi_content(self, move): - #OVERRIDE - if self.code != 'fattura_pa': - return super()._get_invoice_edi_content(move) - return self._export_as_xml(move) - def _check_move_configuration(self, move): # OVERRIDE res = super()._check_move_configuration(move) @@ -840,20 +835,6 @@ class AccountEdiFormat(models.Model): self.ensure_one() return self.code == 'fattura_pa' or super()._needs_web_services() - def _support_batching(self, move=None, state=None, company=None): - # OVERRIDE - if self.code == 'fattura_pa': - return state == 'to_send' and move.is_invoice() - - return super()._support_batching(move=move, state=state, company=company) - - def _get_batch_key(self, move, state): - # OVERRIDE - if self.code != 'fattura_pa': - return super()._get_batch_key(move, state) - - return move.move_type, bool(move.l10n_it_edi_transaction) - def _l10n_it_post_invoices_step_1(self, invoices): ''' Send the invoices to the proxy. ''' @@ -861,7 +842,7 @@ class AccountEdiFormat(models.Model): to_send = {} for invoice in invoices: - xml = "" + str(self._export_as_xml(invoice)) + xml = "" + str(self._l10n_it_edi_export_invoice_as_xml(invoice)) filename = self._l10n_it_edi_generate_electronic_invoice_filename(invoice) attachment = self.env['ir.attachment'].create({ 'name': filename, @@ -1027,12 +1008,12 @@ class AccountEdiFormat(models.Model): return to_return - def _post_fattura_pa(self, invoices): + def _post_fattura_pa(self, invoice): # OVERRIDE - if not invoices[0].l10n_it_edi_transaction: - return self._l10n_it_post_invoices_step_1(invoices) + if not invoice.l10n_it_edi_transaction: + return self._l10n_it_post_invoices_step_1(invoice) else: - return self._l10n_it_post_invoices_step_2(invoices) + return self._l10n_it_post_invoices_step_2(invoice) def _post_invoice_edi(self, invoices): # OVERRIDE diff --git a/addons/l10n_it_edi/models/account_invoice.py b/addons/l10n_it_edi/models/account_invoice.py index 7b738830064..865b389eb22 100644 --- a/addons/l10n_it_edi/models/account_invoice.py +++ b/addons/l10n_it_edi/models/account_invoice.py @@ -43,7 +43,7 @@ class AccountMove(models.Model): self.ensure_one() report_name = self.env['account.edi.format']._l10n_it_edi_generate_electronic_invoice_filename(self) - data = "" + str(self._export_as_xml()) + data = "" + str(self._l10n_it_edi_export_invoice_as_xml()) description = _('Italian invoice: %s', self.move_type) attachment = self.env['ir.attachment'].create({ 'name': report_name, diff --git a/addons/l10n_it_edi/tests/test_edi_export.py b/addons/l10n_it_edi/tests/test_edi_export.py index 259c2daed20..ee866d66396 100644 --- a/addons/l10n_it_edi/tests/test_edi_export.py +++ b/addons/l10n_it_edi/tests/test_edi_export.py @@ -399,7 +399,7 @@ class TestItEdiExport(TestItEdi): 2577.29 ''') - invoice_etree = etree.fromstring(self.edi_format._export_as_xml(self.price_included_invoice)) + invoice_etree = etree.fromstring(self.edi_format._l10n_it_edi_export_invoice_as_xml(self.price_included_invoice)) # Remove the attachment and its details invoice_etree = self.with_applied_xpath(invoice_etree, "") self.assertXmlTreeEqual(invoice_etree, expected_etree) @@ -408,7 +408,7 @@ class TestItEdiExport(TestItEdi): # The EDI can account for discounts, but a line with, for example, a 100% discount should still have # a corresponding tax with a base amount of 0 - invoice_etree = etree.fromstring(self.edi_format._export_as_xml(self.partial_discount_invoice)) + invoice_etree = etree.fromstring(self.edi_format._l10n_it_edi_export_invoice_as_xml(self.partial_discount_invoice)) expected_etree = self.with_applied_xpath( etree.fromstring(self.edi_basis_xml), ''' @@ -465,7 +465,7 @@ class TestItEdiExport(TestItEdi): self.assertXmlTreeEqual(invoice_etree, expected_etree) def test_fully_discounted_inovice(self): - invoice_etree = etree.fromstring(self.edi_format._export_as_xml(self.full_discount_invoice)) + invoice_etree = etree.fromstring(self.edi_format._l10n_it_edi_export_invoice_as_xml(self.full_discount_invoice)) expected_etree = self.with_applied_xpath( etree.fromstring(self.edi_basis_xml), ''' @@ -502,7 +502,7 @@ class TestItEdiExport(TestItEdi): self.assertXmlTreeEqual(invoice_etree, expected_etree) def test_non_latin_and_latin_invoice(self): - invoice_etree = etree.fromstring(self.edi_format._export_as_xml(self.non_latin_and_latin_invoice)) + invoice_etree = etree.fromstring(self.edi_format._l10n_it_edi_export_invoice_as_xml(self.non_latin_and_latin_invoice)) expected_etree = self.with_applied_xpath( etree.fromstring(self.edi_basis_xml), ''' @@ -551,7 +551,7 @@ class TestItEdiExport(TestItEdi): self.assertXmlTreeEqual(invoice_etree, expected_etree) def test_below_400_codice_simplified_invoice(self): - invoice_etree = etree.fromstring(self.edi_format._export_as_xml(self.below_400_codice_simplified_invoice)) + invoice_etree = etree.fromstring(self.edi_format._l10n_it_edi_export_invoice_as_xml(self.below_400_codice_simplified_invoice)) expected_etree = self.with_applied_xpath( etree.fromstring(self.edi_simplified_basis_xml), ''' @@ -581,7 +581,7 @@ class TestItEdiExport(TestItEdi): self.assertXmlTreeEqual(invoice_etree, expected_etree) def test_total_400_VAT_simplified_invoice(self): - invoice_etree = etree.fromstring(self.edi_format._export_as_xml(self.total_400_VAT_simplified_invoice)) + invoice_etree = etree.fromstring(self.edi_format._l10n_it_edi_export_invoice_as_xml(self.total_400_VAT_simplified_invoice)) expected_etree = self.with_applied_xpath( etree.fromstring(self.edi_simplified_basis_xml), ''' @@ -619,7 +619,7 @@ class TestItEdiExport(TestItEdi): self.assertEqual(res[self.pa_partner_invoice], {'attachment': self.pa_partner_invoice.l10n_it_edi_attachment_id, 'success': True}) def test_zero_percent_taxes(self): - invoice_etree = etree.fromstring(self.edi_format._export_as_xml(self.zero_tax_invoice)) + invoice_etree = etree.fromstring(self.edi_format._l10n_it_edi_export_invoice_as_xml(self.zero_tax_invoice)) expected_etree = self.with_applied_xpath( etree.fromstring(self.edi_basis_xml), ''' @@ -667,7 +667,7 @@ class TestItEdiExport(TestItEdi): self.assertXmlTreeEqual(invoice_etree, expected_etree) def test_negative_price_invoice(self): - invoice_etree = etree.fromstring(self.edi_format._export_as_xml(self.negative_price_invoice)) + invoice_etree = etree.fromstring(self.edi_format._l10n_it_edi_export_invoice_as_xml(self.negative_price_invoice)) expected_etree = self.with_applied_xpath( etree.fromstring(self.edi_basis_xml), ''' diff --git a/addons/l10n_it_edi/tests/test_edi_reverse_charge.py b/addons/l10n_it_edi/tests/test_edi_reverse_charge.py index 34b81b624e9..038654dc11a 100644 --- a/addons/l10n_it_edi/tests/test_edi_reverse_charge.py +++ b/addons/l10n_it_edi/tests/test_edi_reverse_charge.py @@ -181,7 +181,7 @@ class TestItEdiReverseCharge(TestItEdi): ) def _test_invoice_with_sample_file(self, invoice, filename, xpaths_file=None, xpaths_result=None): - result = self._cleanup_etree(self.edi_format._export_as_xml(invoice), xpaths_result) + result = self._cleanup_etree(self.edi_format._l10n_it_edi_export_invoice_as_xml(invoice), xpaths_result) expected = self._cleanup_etree(self._get_test_file_content(filename), xpaths_file) self.assertXmlTreeEqual(result, expected)