pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[IMP] account_edi, l10n_*: Make the EDI engine more flexible

Browse Source 
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 <las@odoo.com>
Signed-off-by: Julien Van Roy <juvr@odoo.com>
This commit is contained in:
Laurent Smet
2022-10-04 10:55:34 +02:00
committed by Julien Van Roy
parent 3a96a536e7
commit 150951f61b
13 changed files with 415 additions and 677 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/account_edi/models/account_edi_document.py
+53 -62
View File
@@ -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()
addons/account_edi/models/account_edi_format.py
+9 -109
View File
@@ -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
####################################################
addons/account_edi/models/account_move.py
+31 -22
View File
@@ -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:
addons/account_edi/tests/common.py
+28 -70
View File
@@ -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"<?xml version='1.0' encoding='UTF-8'?><Invoice/>"),
'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"<?xml version='1.0' encoding='UTF-8'?><Invoice/>"),
'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"<?xml version='1.0' encoding='UTF-8'?><Invoice/>"),
'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:
addons/account_edi/tests/test_edi.py
+121 -196
View File
@@ -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())
addons/account_edi_ubl_cii/models/account_edi_format.py
+27 -36
View File
@@ -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
addons/l10n_eg_edi_eta/models/account_edi_format.py
+17 -13
View File
@@ -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
addons/l10n_es_edi_sii/models/account_edi_format.py
+14 -28
View File
@@ -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:
addons/l10n_in_edi/models/account_edi_format.py
+91 -98
View File
@@ -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." \
"<br/>Normally, this should not happen too often" \
"<br/>Just verify value of invoice by uploade json to government website " \
"<a href='https://einvoice1.gst.gov.in/Others/VSignedInvoice'>here<a>."
))
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 = "<br/>".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." \
"<br/>Normally, this should not happen too often" \
"<br/>Just verify by logging into government website " \
"<a href='https://einvoice1.gst.gov.in'>here<a>."
))
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 = "<br/>".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." \
"<br/>Normally, this should not happen too often" \
"<br/>Just verify by logging into government website " \
"<a href='https://einvoice1.gst.gov.in'>here<a>."
))
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 = "<br/>".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()
addons/l10n_it_edi/models/account_edi_format.py
+14 -33
View File
@@ -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 = "<?xml version='1.0' encoding='UTF-8'?>" + str(self._export_as_xml(invoice))
xml = "<?xml version='1.0' encoding='UTF-8'?>" + 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
addons/l10n_it_edi/models/account_invoice.py
+1 -1
View File
@@ -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 = "<?xml version='1.0' encoding='UTF-8'?>" + str(self._export_as_xml())
data = "<?xml version='1.0' encoding='UTF-8'?>" + 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,
addons/l10n_it_edi/tests/test_edi_export.py
+8 -8
View File
@@ -399,7 +399,7 @@ class TestItEdiExport(TestItEdi):
2577.29
</xpath>
''')
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, "<xpath expr='.//Allegati' position='replace'/>")
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),
'''
addons/l10n_it_edi/tests/test_edi_reverse_charge.py
+1 -1
View File
@@ -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)