Odoo18-Base/addons/l10n_hu_edi/tests/test_flows_mocked.py
2025-01-06 10:57:38 +07:00

250 lines
15 KiB
Python

# Part of Odoo. See LICENSE file for full copyright and licensing details.
from odoo import tools
from odoo.tests.common import tagged
from odoo.exceptions import UserError
from odoo.addons.account.tests.test_account_move_send import TestAccountMoveSendCommon
from odoo.addons.l10n_hu_edi.tests.common import L10nHuEdiTestCommon
import requests
from unittest import mock
from freezegun import freeze_time
import contextlib
@tagged('post_install_l10n', '-at_install', 'post_install')
class L10nHuEdiTestFlowsMocked(L10nHuEdiTestCommon, TestAccountMoveSendCommon):
""" Test the Hungarian EDI flows using mocked data from the test servers. """
@classmethod
def setUpClass(cls):
with freeze_time('2024-01-25T15:28:53Z'):
super().setUpClass()
def test_send_invoice_and_credit_note(self):
with self.patch_post(), \
freeze_time('2024-01-25T15:28:53Z'):
invoice = self.create_invoice_simple()
invoice.action_post()
send_and_print = self.create_send_and_print(invoice)
self.assertTrue(send_and_print.extra_edi_checkboxes and send_and_print.extra_edi_checkboxes.get('hu_nav_30', {}).get('checked'))
self.assertFalse(invoice._l10n_hu_edi_check_invoices())
send_and_print.action_send_and_print()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'confirmed', 'l10n_hu_invoice_chain_index': -1}])
credit_note = self.create_reversal(invoice)
credit_note.action_post()
send_and_print = self.create_send_and_print(credit_note)
self.assertTrue(send_and_print.extra_edi_checkboxes and send_and_print.extra_edi_checkboxes.get('hu_nav_30', {}).get('checked'))
self.assertFalse(credit_note._l10n_hu_edi_check_invoices())
send_and_print.action_send_and_print()
self.assertRecordValues(credit_note, [{'l10n_hu_edi_state': 'confirmed', 'l10n_hu_invoice_chain_index': 1}])
def test_send_invoice_warning(self):
with tools.file_open('l10n_hu_edi/tests/mocked_requests/queryTransactionStatus_response_warning.xml', 'r') as response_file:
response_data = response_file.read()
with self.patch_post({'queryTransactionStatus': response_data}), \
freeze_time('2024-01-25T15:28:53Z'):
invoice = self.create_invoice_simple()
invoice.action_post()
send_and_print = self.create_send_and_print(invoice)
self.assertTrue(send_and_print.extra_edi_checkboxes and send_and_print.extra_edi_checkboxes.get('hu_nav_30', {}).get('checked'))
self.assertFalse(invoice._l10n_hu_edi_check_invoices())
send_and_print.action_send_and_print()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'confirmed_warning', 'l10n_hu_invoice_chain_index': -1}])
def test_send_invoice_error(self):
with tools.file_open('l10n_hu_edi/tests/mocked_requests/queryTransactionStatus_response_error.xml', 'r') as response_file:
response_data = response_file.read()
with self.patch_post({'queryTransactionStatus': response_data}), \
freeze_time('2024-01-25T15:28:53Z'):
invoice = self.create_invoice_simple()
invoice.action_post()
send_and_print = self.create_send_and_print(invoice)
self.assertTrue(send_and_print.extra_edi_checkboxes and send_and_print.extra_edi_checkboxes.get('hu_nav_30', {}).get('checked'))
self.assertFalse(invoice._l10n_hu_edi_check_invoices())
with contextlib.suppress(UserError):
send_and_print.action_send_and_print()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'rejected', 'l10n_hu_invoice_chain_index': 0}])
def test_timeout_recovery_fail(self):
with freeze_time('2024-01-25T15:28:53Z'), \
self.patch_post({'manageInvoice': requests.Timeout()}):
invoice = self.create_invoice_simple()
invoice.action_post()
send_and_print = self.create_send_and_print(invoice)
self.assertTrue(send_and_print.extra_edi_checkboxes and send_and_print.extra_edi_checkboxes.get('hu_nav_30', {}).get('checked'))
self.assertFalse(invoice._l10n_hu_edi_check_invoices())
send_and_print.action_send_and_print()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'send_timeout', 'l10n_hu_invoice_chain_index': -1}])
with tools.file_open('l10n_hu_edi/tests/mocked_requests/queryTransactionStatus_response_original.xml', 'r') as response_file:
response_data = response_file.read()
# Advance 10 minutes so the timeout recovery mechanism triggers.
with freeze_time('2024-01-25T15:38:53Z'), \
self.patch_post({'queryTransactionStatus': response_data}):
with contextlib.suppress(UserError):
invoice.l10n_hu_edi_button_update_status()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'rejected', 'l10n_hu_invoice_chain_index': 0}])
def test_timeout_recovery_success(self):
with freeze_time('2024-01-25T15:28:53Z'), \
self.patch_post({'manageInvoice': requests.Timeout()}):
invoice = self.create_invoice_simple()
invoice.name = 'INV/2024/00999' # This matches the invoice name in the XML returned by queryTransactionStatus.
invoice.action_post()
send_and_print = self.create_send_and_print(invoice)
self.assertTrue(send_and_print.extra_edi_checkboxes and send_and_print.extra_edi_checkboxes.get('hu_nav_30', {}).get('checked'))
self.assertFalse(invoice._l10n_hu_edi_check_invoices())
send_and_print.action_send_and_print()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'send_timeout', 'l10n_hu_invoice_chain_index': -1}])
# This returns an original XML with name INV/2024/00999
with tools.file_open('l10n_hu_edi/tests/mocked_requests/queryTransactionStatus_response_original.xml', 'r') as response_file:
response_data = response_file.read()
# Advance 10 minutes so the timeout recovery mechanism triggers.
with freeze_time('2024-01-25T15:38:53Z'), \
self.patch_post({'queryTransactionStatus': response_data}):
invoice.l10n_hu_edi_button_update_status()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'confirmed', 'l10n_hu_invoice_chain_index': -1}])
def test_cancel_invoice_error(self):
with freeze_time('2024-01-25T15:28:53Z'):
with self.patch_post():
invoice, cancel_wizard = self.create_cancel_wizard()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'confirmed'}])
with tools.file_open('l10n_hu_edi/tests/mocked_requests/queryTransactionStatus_response_error.xml', 'r') as response_file:
response_data = response_file.read()
with self.patch_post({'queryTransactionStatus': response_data}):
with contextlib.suppress(UserError):
cancel_wizard.button_request_cancel()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'confirmed_warning', 'l10n_hu_invoice_chain_index': -1}])
def test_cancel_invoice_pending(self):
with freeze_time('2024-01-25T15:28:53Z'):
with self.patch_post():
invoice, cancel_wizard = self.create_cancel_wizard()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'confirmed'}])
with tools.file_open('l10n_hu_edi/tests/mocked_requests/queryTransactionStatus_response_annulment_pending.xml', 'r') as response_file:
response_data = response_file.read()
with self.patch_post({'queryTransactionStatus': response_data}):
cancel_wizard.button_request_cancel()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'cancel_pending', 'l10n_hu_invoice_chain_index': -1}])
def test_cancel_invoice_done(self):
with freeze_time('2024-01-25T15:28:53Z'):
with self.patch_post():
invoice, cancel_wizard = self.create_cancel_wizard()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'confirmed', 'l10n_hu_invoice_chain_index': -1}])
with tools.file_open('l10n_hu_edi/tests/mocked_requests/queryTransactionStatus_response_annulment_done.xml', 'r') as response_file:
response_data = response_file.read()
with self.patch_post({'queryTransactionStatus': response_data}):
cancel_wizard.button_request_cancel()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'cancelled', 'state': 'cancel', 'l10n_hu_invoice_chain_index': 0}])
def test_cancel_and_resend(self):
""" Test the sending, annulment and re-sending of an invoice + credit note + modif. invoice """
with freeze_time('2024-01-25T15:28:53Z'):
with self.patch_post():
invoice, cancel_wizard = self.create_cancel_wizard()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'confirmed', 'l10n_hu_invoice_chain_index': -1}])
new_invoice = self.create_reversal(invoice, is_modify=True)
self.assertRecordValues(new_invoice, [{'debit_origin_id': invoice.id}])
new_invoice.action_post()
credit_note = invoice.reversal_move_ids
send_and_print = self.create_send_and_print(credit_note)
self.assertTrue(send_and_print.extra_edi_checkboxes and send_and_print.extra_edi_checkboxes.get('hu_nav_30', {}).get('checked'))
self.assertFalse(credit_note._l10n_hu_edi_check_invoices())
send_and_print.action_send_and_print()
self.assertRecordValues(credit_note, [{'l10n_hu_edi_state': 'confirmed', 'l10n_hu_invoice_chain_index': 1}])
send_and_print = self.create_send_and_print(new_invoice)
self.assertTrue(send_and_print.extra_edi_checkboxes and send_and_print.extra_edi_checkboxes.get('hu_nav_30', {}).get('checked'))
self.assertFalse(new_invoice._l10n_hu_edi_check_invoices())
send_and_print.action_send_and_print()
self.assertRecordValues(new_invoice, [{'l10n_hu_edi_state': 'confirmed', 'l10n_hu_invoice_chain_index': 2}])
with tools.file_open('l10n_hu_edi/tests/mocked_requests/queryTransactionStatus_response_annulment_done.xml', 'r') as response_file:
response_data = response_file.read()
with self.patch_post({'queryTransactionStatus': response_data}):
cancel_wizard.button_request_cancel()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'cancelled', 'state': 'cancel', 'l10n_hu_invoice_chain_index': 0}])
self.assertRecordValues(credit_note, [{'l10n_hu_edi_state': 'cancelled', 'state': 'cancel', 'l10n_hu_invoice_chain_index': 0}])
self.assertRecordValues(new_invoice, [{'l10n_hu_edi_state': 'cancelled', 'state': 'cancel', 'l10n_hu_invoice_chain_index': 0}])
(invoice | credit_note | new_invoice).button_draft()
invoice.action_post()
credit_note.action_post()
new_invoice.action_post()
with self.patch_post():
send_and_print = self.create_send_and_print(invoice)
self.assertTrue(send_and_print.extra_edi_checkboxes and send_and_print.extra_edi_checkboxes.get('hu_nav_30', {}).get('checked'))
self.assertFalse(invoice._l10n_hu_edi_check_invoices())
send_and_print.action_send_and_print()
self.assertRecordValues(invoice, [{'l10n_hu_edi_state': 'confirmed', 'l10n_hu_invoice_chain_index': -1}])
send_and_print = self.create_send_and_print(credit_note)
self.assertTrue(send_and_print.extra_edi_checkboxes and send_and_print.extra_edi_checkboxes.get('hu_nav_30', {}).get('checked'))
self.assertFalse(credit_note._l10n_hu_edi_check_invoices())
send_and_print.action_send_and_print()
self.assertRecordValues(credit_note, [{'l10n_hu_edi_state': 'confirmed', 'l10n_hu_invoice_chain_index': 1}])
send_and_print = self.create_send_and_print(new_invoice)
self.assertTrue(send_and_print.extra_edi_checkboxes and send_and_print.extra_edi_checkboxes.get('hu_nav_30', {}).get('checked'))
self.assertFalse(new_invoice._l10n_hu_edi_check_invoices())
send_and_print.action_send_and_print()
self.assertRecordValues(new_invoice, [{'l10n_hu_edi_state': 'confirmed', 'l10n_hu_invoice_chain_index': 2}])
# === Helpers === #
@contextlib.contextmanager
def patch_post(self, responses=None):
""" Patch requests.Session in l10n_hu_edi.connection.
:param responses: If specified, a dict {service: response} that gives, for any service,
bytes that should be served as response data, or an Exception that should be raised.
Otherwise, will use the default responses stored under
mocked_requests/{service}_response.xml
"""
test_case = self
class MockedSession:
def post(self, url, data, headers, timeout=None):
prod_url = 'https://api.onlineszamla.nav.gov.hu/invoiceService/v3'
demo_url = 'https://api-test.onlineszamla.nav.gov.hu/invoiceService/v3'
mocked_requests = ['manageInvoice', 'queryTaxpayer', 'tokenExchange', 'queryTransactionStatus', 'queryTransactionList', 'manageAnnulment']
base_url, __, service = url.rpartition('/')
if base_url not in (prod_url, demo_url) or service not in mocked_requests:
test_case.fail(f'Invalid POST url: {url}')
with tools.file_open(f'l10n_hu_edi/tests/mocked_requests/{service}_request.xml', 'rb') as expected_request_file:
test_case.assertXmlTreeEqual(
test_case.get_xml_tree_from_string(data),
test_case.get_xml_tree_from_string(expected_request_file.read()),
)
mock_response = mock.Mock(spec=requests.Response)
mock_response.status_code = 200
mock_response.headers = ''
if responses and service in responses:
if isinstance(responses[service], Exception):
raise responses[service]
mock_response.text = responses[service]
else:
with tools.file_open(f'l10n_hu_edi/tests/mocked_requests/{service}_response.xml', 'r') as response_file:
mock_response.text = response_file.read()
return mock_response
def close(self):
pass
with mock.patch('odoo.addons.l10n_hu_edi.models.l10n_hu_edi_connection.requests.Session', side_effect=MockedSession, autospec=True):
yield