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

[FIX] account_peppol: mock external request

Browse Source 
Now that we check whether a participant exists on Peppol network first when they register,
we should also mock that request. Currently, since that request is not mocked,
all peppol tests are skipped.

Part-of: odoo/odoo#158546
This commit is contained in:
aliya
2024-03-22 13:02:02 +00:00
parent 57225048f9
commit bde46a0b17
2 changed files with 159 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/account_peppol/tests/test_peppol_messages.py
+55 -52
View File
@@ -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',
}])
addons/account_peppol/tests/test_peppol_participant.py
+104 -93
View File
@@ -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'<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n<smp:ServiceGroup xmlns:wsa="http://www.w3.org/2005/08/addressing" xmlns:id="http://busdox.org/transport/identifiers/1.0/" xmlns:ds="http://www.w3.org/2000/09/xmldsig#" xmlns:smp="http://busdox.org/serviceMetadata/publishing/1.0/"><id:ParticipantIdentifier scheme="iso6523-actorid-upis">0208:0000000000</id:ParticipantIdentifier></smp:ServiceGroup>'
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: