mirror of
https://github.com/bringout/oca-ocb-accounting.git
synced 2026-04-21 10:02:02 +02:00
Initial commit: Accounting packages
This commit is contained in:
Ernad Husremovic
commit
4ef34c2317
2661 changed files with 1709616 additions and 0 deletions
|
|
@ -0,0 +1,6 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
||||
|
||||
from . import common
|
||||
from . import test_edi
|
||||
from . import test_import_vendor_bill
|
||||
175
odoo-bringout-oca-ocb-account_edi/account_edi/tests/common.py
Normal file
175
odoo-bringout-oca-ocb-account_edi/account_edi/tests/common.py
Normal file
|
|
@ -0,0 +1,175 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
||||
from odoo.modules.module import get_module_resource
|
||||
from odoo.addons.account.tests.common import AccountTestInvoicingCommon
|
||||
|
||||
from contextlib import contextmanager
|
||||
from unittest.mock import patch
|
||||
|
||||
import base64
|
||||
|
||||
|
||||
def _generate_mocked_needs_web_services(needs_web_services):
|
||||
return lambda edi_format: needs_web_services
|
||||
|
||||
|
||||
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):
|
||||
return []
|
||||
|
||||
|
||||
def _mocked_check_move_configuration_fail(edi_format, move):
|
||||
return ['Fake error (mocked)']
|
||||
|
||||
|
||||
def _mocked_cancel_success(edi_format, invoices):
|
||||
return {invoice: {'success': True} for invoice in invoices}
|
||||
|
||||
|
||||
class AccountEdiTestCommon(AccountTestInvoicingCommon):
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls, chart_template_ref=None, edi_format_ref=None):
|
||||
super().setUpClass(chart_template_ref=chart_template_ref)
|
||||
|
||||
# ==== EDI ====
|
||||
if edi_format_ref:
|
||||
cls.edi_format = cls.env.ref(edi_format_ref)
|
||||
else:
|
||||
with cls.mock_edi(cls, _needs_web_services_method=_generate_mocked_needs_web_services(True)):
|
||||
cls.edi_format = cls.env['account.edi.format'].sudo().create({
|
||||
'name': 'Test EDI format',
|
||||
'code': 'test_edi',
|
||||
})
|
||||
cls.journal = cls.company_data['default_journal_sale']
|
||||
cls.journal.edi_format_ids = [(6, 0, cls.edi_format.ids)]
|
||||
|
||||
####################################################
|
||||
# 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,
|
||||
_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,
|
||||
):
|
||||
|
||||
try:
|
||||
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._check_move_configuration',
|
||||
new=_check_move_configuration_method), \
|
||||
patch('odoo.addons.account_edi.models.account_edi_format.AccountEdiFormat._get_move_applicability',
|
||||
new=_get_move_applicability_method):
|
||||
|
||||
yield
|
||||
finally:
|
||||
pass
|
||||
|
||||
def edi_cron(self):
|
||||
self.env['account.edi.document'].sudo().search([('state', 'in', ('to_send', 'to_cancel'))])._process_documents_web_services(with_commit=False)
|
||||
|
||||
def _create_empty_vendor_bill(self):
|
||||
invoice = self.env['account.move'].create({
|
||||
'move_type': 'in_invoice',
|
||||
'journal_id': self.company_data['default_journal_purchase'].id,
|
||||
})
|
||||
return invoice
|
||||
|
||||
def update_invoice_from_file(self, module_name, subfolder, filename, invoice):
|
||||
file_path = get_module_resource(module_name, subfolder, filename)
|
||||
file = open(file_path, 'rb').read()
|
||||
|
||||
attachment = self.env['ir.attachment'].create({
|
||||
'name': filename,
|
||||
'datas': base64.encodebytes(file),
|
||||
'res_id': invoice.id,
|
||||
'res_model': 'account.move',
|
||||
})
|
||||
|
||||
invoice.message_post(attachment_ids=[attachment.id])
|
||||
|
||||
def create_invoice_from_file(self, module_name, subfolder, filename):
|
||||
file_path = get_module_resource(module_name, subfolder, filename)
|
||||
file = open(file_path, 'rb').read()
|
||||
|
||||
attachment = self.env['ir.attachment'].create({
|
||||
'name': filename,
|
||||
'datas': base64.encodebytes(file),
|
||||
'res_model': 'account.move',
|
||||
})
|
||||
journal_id = self.company_data['default_journal_sale']
|
||||
action_vals = journal_id.with_context(default_move_type='in_invoice').create_document_from_attachment(attachment.ids)
|
||||
return self.env['account.move'].browse(action_vals['res_id'])
|
||||
|
||||
def assert_generated_file_equal(self, invoice, expected_values, applied_xpath=None):
|
||||
invoice.action_post()
|
||||
invoice.edi_document_ids._process_documents_web_services(with_commit=False) # synchronous are called in post, but there's no CRON in tests for asynchronous
|
||||
attachment = invoice._get_edi_attachment(self.edi_format)
|
||||
if not attachment:
|
||||
raise ValueError('No attachment was generated after posting EDI')
|
||||
xml_content = base64.b64decode(attachment.with_context(bin_size=False).datas)
|
||||
current_etree = self.get_xml_tree_from_string(xml_content)
|
||||
expected_etree = self.get_xml_tree_from_string(expected_values)
|
||||
if applied_xpath:
|
||||
expected_etree = self.with_applied_xpath(expected_etree, applied_xpath)
|
||||
self.assertXmlTreeEqual(current_etree, expected_etree)
|
||||
|
||||
def create_edi_document(self, edi_format, state, move=None, move_type=None):
|
||||
""" Creates a document based on an existing invoice or creates one, too.
|
||||
|
||||
:param edi_format: The edi_format of the document.
|
||||
:param state: The state of the document.
|
||||
:param move: The move of the document or None to create a new one.
|
||||
:param move_type: If move is None, the type of the invoice to create, defaults to 'out_invoice'.
|
||||
"""
|
||||
move = move or self.init_invoice(move_type or 'out_invoice', products=self.product_a)
|
||||
return self.env['account.edi.document'].create({
|
||||
'edi_format_id': edi_format.id,
|
||||
'move_id': move.id,
|
||||
'state': state
|
||||
})
|
||||
|
||||
def _process_documents_web_services(self, moves, formats_to_return=None):
|
||||
""" Generates and returns EDI files for the specified moves.
|
||||
formats_to_return is an optional parameter used to pass a set of codes from
|
||||
the formats we want to return the files for (in case we want to test specific formats).
|
||||
Other formats will still generate documents, they simply won't be returned.
|
||||
"""
|
||||
moves.edi_document_ids._process_documents_web_services(with_commit=False)
|
||||
|
||||
documents_to_return = moves.edi_document_ids
|
||||
if formats_to_return != None:
|
||||
documents_to_return = documents_to_return.filtered(lambda x: x.edi_format_id.code in formats_to_return)
|
||||
|
||||
attachments = documents_to_return.sudo().attachment_id
|
||||
data_str_list = []
|
||||
for attachment in attachments.with_context(bin_size=False):
|
||||
data_str_list.append(base64.decodebytes(attachment.datas))
|
||||
return data_str_list
|
||||
170
odoo-bringout-oca-ocb-account_edi/account_edi/tests/test_edi.py
Normal file
170
odoo-bringout-oca-ocb-account_edi/account_edi/tests/test_edi.py
Normal file
|
|
@ -0,0 +1,170 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
||||
|
||||
from odoo.addons.account_edi.tests.common import AccountEdiTestCommon
|
||||
from odoo.addons.base.tests.test_ir_cron import CronMixinCase
|
||||
from odoo.tests import tagged
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install')
|
||||
class TestAccountEdi(AccountEdiTestCommon, CronMixinCase):
|
||||
|
||||
@classmethod
|
||||
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.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):
|
||||
invoice = self.init_invoice('out_invoice', products=self.product_a)
|
||||
|
||||
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)
|
||||
|
||||
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)
|
||||
|
||||
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()
|
||||
|
||||
jobs = (invoice1 + invoice2).edi_document_ids._prepare_jobs()
|
||||
self.assertEqual(len(jobs), 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)
|
||||
|
||||
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()
|
||||
|
||||
jobs = (invoice1 + invoice2).edi_document_ids._prepare_jobs()
|
||||
self.assertEqual(len(jobs), 1)
|
||||
|
||||
def test_warning_is_retried(self):
|
||||
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):
|
||||
|
||||
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):
|
||||
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):
|
||||
def step1(edi_format, invoice):
|
||||
return {invoice: {'error': "step1 done", 'blocking_level': 'info'}}
|
||||
|
||||
def step2(edi_format, invoice):
|
||||
return {invoice: {'success': True}}
|
||||
|
||||
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:
|
||||
return {'post': edi_format._test_edi_post_invoice_step1}
|
||||
|
||||
invoice = self.init_invoice('out_invoice', products=self.product_a)
|
||||
|
||||
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'}])
|
||||
|
||||
invoice.action_process_edi_web_services(with_commit=False)
|
||||
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'}])
|
||||
|
||||
def test_cron_triggers(self):
|
||||
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):
|
||||
# Process single job by CRON call (and thus, disable the auto-commit).
|
||||
edi_cron = self.env.ref('account_edi.ir_cron_edi_network')
|
||||
edi_cron.code = 'model._cron_process_documents_web_services(job_count=1)'
|
||||
|
||||
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",
|
||||
)
|
||||
|
||||
def test_invoice_ready_to_be_sent(self):
|
||||
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())
|
||||
|
|
@ -0,0 +1,21 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
from odoo.addons.account.tests.common import AccountTestInvoicingCommon
|
||||
from odoo.tests import tagged
|
||||
|
||||
|
||||
@tagged('post_install', '-at_install')
|
||||
class TestImportVendorBill(AccountTestInvoicingCommon):
|
||||
|
||||
def test_retrieve_partner(self):
|
||||
|
||||
def retrieve_partner(vat, import_vat):
|
||||
self.partner_a.with_context(no_vat_validation=True).vat = vat
|
||||
return self.env['account.edi.format']._retrieve_partner(vat=import_vat)
|
||||
|
||||
self.assertEqual(self.partner_a, retrieve_partner('BE0477472701', 'BE0477472701'))
|
||||
self.assertEqual(self.partner_a, retrieve_partner('BE0477472701', '0477472701'))
|
||||
self.assertEqual(self.partner_a, retrieve_partner('BE0477472701', '477472701'))
|
||||
self.assertEqual(self.partner_a, retrieve_partner('0477472701', 'BE0477472701'))
|
||||
self.assertEqual(self.partner_a, retrieve_partner('477472701', 'BE0477472701'))
|
||||
self.assertEqual(self.env['res.partner'], retrieve_partner('DE0477472701', 'BE0477472701'))
|
||||
self.assertEqual(self.partner_a, retrieve_partner('CHE-107.787.577 IVA', 'CHE-107.787.577 IVA')) # note that base_vat forces the space
|
||||
Loading…
Add table
Add a link
Reference in a new issue