diff --git a/addons/account_peppol/tests/test_peppol_messages.py b/addons/account_peppol/tests/test_peppol_messages.py index f5043653d51..5023c95fe9b 100644 --- a/addons/account_peppol/tests/test_peppol_messages.py +++ b/addons/account_peppol/tests/test_peppol_messages.py @@ -1,9 +1,8 @@ -# -*- coding: utf-8 -* - +import json from base64 import b64encode from contextlib import contextmanager from freezegun import freeze_time -from unittest.mock import Mock, patch +from requests import Session, PreparedRequest, Response from odoo.addons.account.tests.test_account_move_send import TestAccountMoveSendCommon from odoo.exceptions import UserError @@ -16,7 +15,6 @@ FAKE_UUID = ['yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy', 'zzzzzzzz-zzzz-zzzz-zzzz-zzzzzzzzzzzz'] FILE_PATH = 'account_peppol/tests/assets' - @freeze_time('2023-01-01') @tagged('-at_install', 'post_install') class TestPeppolMessage(TestAccountMoveSendCommon): @@ -83,8 +81,8 @@ class TestPeppolMessage(TestAccountMoveSendCommon): ], }) - @contextmanager - def _patch_peppol_requests(self, error=False): + @classmethod + def _get_mock_data(cls, error=False): proxy_documents = { FAKE_UUID[0]: { 'accounting_supplier_party': False, @@ -107,10 +105,10 @@ class TestPeppolMessage(TestAccountMoveSendCommon): } responses = { - '/peppol/1/send_document': {'result': { + '/api/peppol/1/send_document': {'result': { 'messages': [{'message_uuid': FAKE_UUID[0]}]}}, - '/peppol/1/ack': {'result': {}}, - '/peppol/1/get_all_documents': {'result': { + '/api/peppol/1/ack': {'result': {}}, + '/api/peppol/1/get_all_documents': {'result': { 'messages': [ { 'accounting_supplier_party': '0208:2718281828', @@ -127,29 +125,38 @@ class TestPeppolMessage(TestAccountMoveSendCommon): ], }} } + return proxy_documents, responses - def _mocked_post(url, *args, **kwargs): - response = Mock() - response.status_code = 200 - url = url.split('api')[1] - if url.endswith('api/peppol/1/send_document'): - if not kwargs['json']['params']['documents']: - raise UserError('No documents were provided') - - if url == '/peppol/1/get_document': - uuid = kwargs['json']['params']['message_uuids'][0] - response.json = lambda: { - 'result': {uuid: proxy_documents[uuid]}} - return response - - if url not in responses: - raise Exception(f'Unexpected request: {url}') - response.json = lambda: responses[url] + @contextmanager + def _set_context(self, other_context): + previous_context = self.env.context + self.env.context = dict(previous_context, **other_context) + yield self + self.env.context = previous_context + @classmethod + def _request_handler(cls, s: Session, r: PreparedRequest, /, **kw): + response = Response() + response.status_code = 200 + if r.url.endswith('/iso6523-actorid-upis%3A%3A9925%3A0000000000'): return response - with patch('odoo.addons.account_edi_proxy_client.models.account_edi_proxy_user.requests.post', side_effect=_mocked_post): - yield + proxy_documents, responses = cls._get_mock_data(cls.env.context.get('error')) + url = r.path_url + body = json.loads(r.body) + if url == '/api/peppol/1/send_document': + if not body['params']['documents']: + raise UserError('No documents were provided') + + if url == '/api/peppol/1/get_document': + uuid = body['params']['message_uuids'][0] + response.json = lambda: {'result': {uuid: proxy_documents[uuid]}} + return response + + if url not in responses: + return super()._request_handler(s, r, **kw) + response.json = lambda: responses[url] + return response def test_attachment_placeholders(self): move = self.create_move(self.valid_partner) @@ -206,8 +213,7 @@ class TestPeppolMessage(TestAccountMoveSendCommon): checkbox_ubl_cii_xml=True, checkbox_send_peppol=True, ) - - with self._patch_peppol_requests(error=True): + with self._set_context({'error': True}): wizard.action_send_and_print() self.env['account_edi_proxy_client.user']._cron_peppol_get_message_status() @@ -225,11 +231,10 @@ class TestPeppolMessage(TestAccountMoveSendCommon): checkbox_send_peppol=True, ) - with self._patch_peppol_requests(): - wizard.action_send_and_print() + wizard.action_send_and_print() - self.env['account_edi_proxy_client.user']._cron_peppol_get_message_status() - self.assertEqual(move.peppol_move_state, 'done') + self.env['account_edi_proxy_client.user']._cron_peppol_get_message_status() + self.assertEqual(move.peppol_move_state, 'done') def test_send_success_message(self): # should be able to send valid invoices correctly @@ -244,16 +249,15 @@ class TestPeppolMessage(TestAccountMoveSendCommon): checkbox_send_peppol=True, ) - with self._patch_peppol_requests(): - wizard.action_send_and_print() + wizard.action_send_and_print() - self.env['account_edi_proxy_client.user']._cron_peppol_get_message_status() - self.assertRecordValues( - move, [{ - 'peppol_move_state': 'done', - 'peppol_message_uuid': FAKE_UUID[0], - }]) - self.assertTrue(bool(move.ubl_cii_xml_id)) + self.env['account_edi_proxy_client.user']._cron_peppol_get_message_status() + self.assertRecordValues(move, [{ + 'peppol_move_state': 'done', + 'peppol_message_uuid': FAKE_UUID[0], + }], + ) + self.assertTrue(bool(move.ubl_cii_xml_id)) def test_send_invalid_edi_user(self): # an invalid edi user should not be able to send invoices via peppol @@ -270,7 +274,7 @@ class TestPeppolMessage(TestAccountMoveSendCommon): def test_receive_error_peppol(self): # an error peppol message should be created - with self._patch_peppol_requests(error=True): + with self._set_context({'error': True}): self.env['account_edi_proxy_client.user']._cron_peppol_get_new_documents() move = self.env['account.move'].search([('peppol_message_uuid', '=', FAKE_UUID[1])]) @@ -282,12 +286,11 @@ class TestPeppolMessage(TestAccountMoveSendCommon): def test_receive_success_peppol(self): # a correct move should be created - with self._patch_peppol_requests(): - self.env['account_edi_proxy_client.user']._cron_peppol_get_new_documents() + self.env['account_edi_proxy_client.user']._cron_peppol_get_new_documents() - move = self.env['account.move'].search([('peppol_message_uuid', '=', FAKE_UUID[1])]) - self.assertRecordValues( - move, [{ - 'peppol_move_state': 'done', - 'move_type': 'in_invoice', - }]) + move = self.env['account.move'].search([('peppol_message_uuid', '=', FAKE_UUID[1])]) + self.assertRecordValues( + move, [{ + 'peppol_move_state': 'done', + 'move_type': 'in_invoice', + }]) diff --git a/addons/account_peppol/tests/test_peppol_participant.py b/addons/account_peppol/tests/test_peppol_participant.py index 88561ddb54d..3f32654f1bd 100644 --- a/addons/account_peppol/tests/test_peppol_participant.py +++ b/addons/account_peppol/tests/test_peppol_participant.py @@ -1,8 +1,7 @@ -# -*- coding: utf-8 -* - +import json from contextlib import contextmanager from freezegun import freeze_time -from unittest.mock import Mock, patch +from requests import Session, PreparedRequest, Response from psycopg2 import IntegrityError from odoo.exceptions import ValidationError, UserError @@ -13,7 +12,6 @@ ID_CLIENT = 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx' FAKE_UUID = 'yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy' PDF_FILE_PATH = 'account_peppol/tests/assets/peppol_identification_test.pdf' - @freeze_time('2023-01-01') @tagged('-at_install', 'post_install') class TestPeppolParticipant(TransactionCase): @@ -23,6 +21,69 @@ class TestPeppolParticipant(TransactionCase): super().setUpClass() cls.env['ir.config_parameter'].sudo().set_param('account_peppol.edi.mode', 'test') + @classmethod + def _get_mock_responses(cls, reject=False): + return { + '/api/peppol/1/participant_status': { + 'result': { + 'peppol_state': 'active' if not reject else 'rejected', + } + }, + '/api/peppol/1/activate_participant': {'result': {}}, + '/iap/account_edi/2/create_user': { + 'result': { + 'id_client': ID_CLIENT, + 'refresh_token': FAKE_UUID, + } + }, + '/api/peppol/1/send_verification_code': {'result': {}}, + '/api/peppol/1/update_user': {'result': {}}, + '/api/peppol/1/verify_phone_number': {'result': {}}, + '/api/peppol/1/migrate_peppol_registration': { + 'result': { + 'migration_key': 'test_key', + } + }, + } + + @classmethod + def _request_handler(cls, s: Session, r: PreparedRequest, /, **kw): + response = Response() + response.status_code = 200 + if r.url.endswith('/iso6523-actorid-upis%3A%3A9925%3A0000000000'): + response.status_code = 404 + return response + + if r.url.endswith('/iso6523-actorid-upis%3A%3A0208%3A0000000000'): + response._content = b'\n0208:0000000000' + return response + + url = r.path_url + body = json.loads(r.body) + responses = cls._get_mock_responses(cls.env.context.get('reject')) + if ( + url == '/api/peppol/1/activate_participant' + and cls.env.context.get('migrate_to') + and not body['params']['migration_key'] + ): + raise UserError('No migration key was provided') + + if cls.env.context.get('migrated_away'): + response.json = lambda: { + 'result': { + 'proxy_error': { + 'code': 'no_such_user', + 'message': 'The user does not exist on the proxy', + } + } + } + return response + + if url not in responses: + return super()._request_handler(s, r, **kw) + response.json = lambda: responses[url] + return response + def _get_participant_vals(self): return { 'account_peppol_eas': '9925', @@ -32,60 +93,11 @@ class TestPeppolParticipant(TransactionCase): } @contextmanager - def _patch_peppol_requests(self, reject=False, migrate_to=False, migrated_away=False): - responses = { - '/peppol/1/participant_status': { - 'result': { - 'peppol_state': 'active' if not reject else 'rejected', - } - }, - '/peppol/1/activate_participant': {'result': {}}, - '/account_edi/2/create_user': { - 'result': { - 'id_client': ID_CLIENT, - 'refresh_token': FAKE_UUID, - } - }, - '/peppol/1/send_verification_code': {'result': {}}, - '/peppol/1/update_user': {'result': {}}, - '/peppol/1/verify_phone_number': {'result': {}}, - '/peppol/1/migrate_peppol_registration': { - 'result': { - 'migration_key': 'test_key', - } - }, - } - - def _mocked_post(url, *args, **kwargs): - response = Mock() - response.status_code = 200 - if ( - url.endswith('/api/peppol/1/activate_participant') - and migrate_to - and not kwargs['json']['params']['migration_key'] - ): - raise UserError('No migration key was provided') - - if migrated_away: - response.json = lambda: { - 'result': { - 'proxy_error': { - 'code': 'no_such_user', - 'message': 'The user does not exist on the proxy', - } - } - } - return response - - url = url.split('/api')[1] if 'iap' not in url else url.split('/iap')[1] - if url not in responses: - raise Exception(f'Unexpected request: {url}') - response.json = lambda: responses[url] - - return response - - with patch('odoo.addons.account_edi_proxy_client.models.account_edi_proxy_user.requests.post', side_effect=_mocked_post): - yield + def _set_context(self, other_context): + previous_context = self.env.context + self.env.context = dict(previous_context, **other_context) + yield self + self.env.context = previous_context def test_create_participant_missing_data(self): # creating a participant without eas/endpoint/document should not be possible @@ -96,6 +108,14 @@ class TestPeppolParticipant(TransactionCase): with self.assertRaises(ValidationError), self.cr.savepoint(): settings.button_create_peppol_proxy_user() + def test_create_participant_already_exists(self): + # creating a participant that already exists on Peppol network should not be possible + vals = self._get_participant_vals() + vals['account_peppol_eas'] = '0208' + settings = self.env['res.config.settings'].create(vals) + with self.assertRaises(UserError), self.cr.savepoint(): + settings.button_create_peppol_proxy_user() + def test_create_success_participant(self): # should be possible to apply with all data # the account_peppol_proxy_state should correctly change to pending @@ -103,17 +123,15 @@ class TestPeppolParticipant(TransactionCase): # after checking participant status company = self.env.company settings = self.env['res.config.settings'].create(self._get_participant_vals()) - - with self._patch_peppol_requests(): - settings.button_create_peppol_proxy_user() - self.assertEqual(company.account_peppol_proxy_state, 'not_verified') - settings.button_send_peppol_verification_code() - self.assertEqual(company.account_peppol_proxy_state, 'sent_verification') - settings.account_peppol_verification_code = '123456' - settings.button_check_peppol_verification_code() - self.assertEqual(company.account_peppol_proxy_state, 'pending') - self.env['account_edi_proxy_client.user']._cron_peppol_get_participant_status() - self.assertEqual(company.account_peppol_proxy_state, 'active') + settings.button_create_peppol_proxy_user() + self.assertEqual(company.account_peppol_proxy_state, 'not_verified') + settings.button_send_peppol_verification_code() + self.assertEqual(company.account_peppol_proxy_state, 'sent_verification') + settings.account_peppol_verification_code = '123456' + settings.button_check_peppol_verification_code() + self.assertEqual(company.account_peppol_proxy_state, 'pending') + self.env['account_edi_proxy_client.user']._cron_peppol_get_participant_status() + self.assertEqual(company.account_peppol_proxy_state, 'active') def test_create_reject_participant(self): # the account_peppol_proxy_state should change to rejected @@ -121,7 +139,7 @@ class TestPeppolParticipant(TransactionCase): company = self.env.company settings = self.env['res.config.settings'].create(self._get_participant_vals()) - with self._patch_peppol_requests(reject=True): + with self._set_context({'reject': True}): settings.button_create_peppol_proxy_user() company.account_peppol_proxy_state = 'pending' self.env['account_edi_proxy_client.user']._cron_peppol_get_participant_status() @@ -131,12 +149,10 @@ class TestPeppolParticipant(TransactionCase): def test_create_duplicate_participant(self): # should not be possible to create a duplicate participant settings = self.env['res.config.settings'].create(self._get_participant_vals()) - - with self._patch_peppol_requests(): + settings.button_create_peppol_proxy_user() + with self.assertRaises(IntegrityError), self.cr.savepoint(): + settings.account_peppol_proxy_state = 'not_registered' settings.button_create_peppol_proxy_user() - with self.assertRaises(IntegrityError), self.cr.savepoint(): - settings.account_peppol_proxy_state = 'not_registered' - settings.button_create_peppol_proxy_user() def test_save_migration_key(self): # migration key should be saved @@ -146,7 +162,7 @@ class TestPeppolParticipant(TransactionCase): 'account_peppol_migration_key': 'helloo', }) - with self._patch_peppol_requests(migrate_to=True): + with self._set_context({'migrate_to': True}): settings.button_create_peppol_proxy_user() self.assertEqual(self.env.company.account_peppol_proxy_state, 'not_verified') self.assertFalse(settings.account_peppol_migration_key) # the key should be reset once we've used it @@ -154,34 +170,29 @@ class TestPeppolParticipant(TransactionCase): def test_migrate_away_participant(self): # a participant should be able to request a migration key settings = self.env['res.config.settings'].create(self._get_participant_vals()) - - with self._patch_peppol_requests(): - self.assertFalse(settings.account_peppol_migration_key) - settings.button_create_peppol_proxy_user() - settings.account_peppol_proxy_state = 'active' - settings.button_migrate_peppol_registration() - self.assertEqual(self.env.company.account_peppol_proxy_state, 'active') - self.assertEqual(settings.account_peppol_migration_key, 'test_key') + self.assertFalse(settings.account_peppol_migration_key) + settings.button_create_peppol_proxy_user() + settings.account_peppol_proxy_state = 'active' + settings.button_migrate_peppol_registration() + self.assertEqual(self.env.company.account_peppol_proxy_state, 'active') + self.assertEqual(settings.account_peppol_migration_key, 'test_key') def test_reset_participant(self): # once a participant has migrated away, they should be reset settings = self.env['res.config.settings'].create(self._get_participant_vals()) + settings.button_create_peppol_proxy_user() + settings.account_peppol_proxy_state = 'active' + settings.button_migrate_peppol_registration() - with self._patch_peppol_requests(): - settings.button_create_peppol_proxy_user() - settings.account_peppol_proxy_state = 'active' - settings.button_migrate_peppol_registration() - - with self._patch_peppol_requests(migrated_away=True): + with self._set_context({'migrated_away': True}): try: settings.button_update_peppol_user_data() except UserError: settings = self.env['res.config.settings'].create({}) - self.assertRecordValues( - settings, [{ + self.assertRecordValues(settings, [{ 'account_peppol_migration_key': False, 'account_peppol_proxy_state': 'not_registered', - }] + }], ) self.assertFalse(self.env.company.account_edi_proxy_client_ids.filtered(lambda u: u.proxy_type == 'peppol')) else: