1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247
|
import json
from contextlib import contextmanager
from requests import Session, PreparedRequest, Response
from psycopg2 import IntegrityError
from odoo.exceptions import ValidationError, UserError
from odoo.tests.common import tagged, TransactionCase, freeze_time
from odoo.tools import mute_logger
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):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.env['ir.config_parameter'].sudo().set_param('account_peppol.edi.mode', 'test')
@classmethod
def _get_mock_responses(cls):
participant_state = cls.env.context.get('participant_state', 'receiver')
return {
'/api/peppol/2/participant_status': {
'result': {
'peppol_state': participant_state,
}
},
'/api/peppol/1/activate_participant': {'result': {}},
'/api/peppol/2/register_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/2/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()
if (
url == '/api/peppol/2/register_participant'
and cls.env.context.get('migrate_to')
and not body['params'].get('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 {
'peppol_eas': '9925',
'peppol_endpoint': '0000000000',
'phone_number': '+32483123456',
'contact_email': 'yourcompany@test.example.com',
}
@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
def test_create_participant_missing_data(self):
# creating a participant without eas/endpoint/document should not be possible
wizard = self.env['peppol.registration'].create({
'peppol_eas': False,
'peppol_endpoint': False,
})
with self.assertRaises(ValidationError), self.cr.savepoint():
wizard.button_peppol_sender_registration()
def test_create_participant_already_exists(self):
# creating a receiver participant that already exists on Peppol network should not be possible
vals = self._get_participant_vals()
vals['peppol_eas'] = '0208'
wizard = self.env['peppol.registration'].create(vals)
wizard.smp_registration = True
with self.assertRaises(UserError), self.cr.savepoint():
wizard.button_peppol_sender_registration()
wizard.verification_code = '123456'
wizard.button_check_peppol_verification_code()
def test_create_success_sender(self):
# should be possible to apply with all data
# the account_peppol_proxy_state should correctly change to sender
# then the account_peppol_proxy_state should not change
# after running the cron checking participant status
company = self.env.company
wizard = self.env['peppol.registration'].create(self._get_participant_vals())
wizard.button_peppol_sender_registration()
# should send verification code immediately
self.assertEqual(company.account_peppol_proxy_state, 'in_verification')
wizard.verification_code = '123456'
wizard.button_check_peppol_verification_code()
# since we did not select receiver registration, we're now just a sender
self.assertEqual(company.account_peppol_proxy_state, 'sender')
# running the cron should not do anything for the company
with self._set_context({'participant_state': 'sender'}):
self.env['account_edi_proxy_client.user']._cron_peppol_get_participant_status()
self.assertEqual(company.account_peppol_proxy_state, 'sender')
def test_create_success_receiver(self):
# should be possible to apply with all data
# the account_peppol_proxy_state should correctly change to smp_registration
# then the account_peppol_proxy_state should change successfully
# after checking participant status
company = self.env.company
wizard = self.env['peppol.registration'].create(self._get_participant_vals())
wizard.smp_registration = True # choose to register as a receiver right away
wizard.button_peppol_sender_registration()
# should send verification code immediately
self.assertEqual(company.account_peppol_proxy_state, 'in_verification')
wizard.verification_code = '123456'
wizard.button_check_peppol_verification_code()
self.assertEqual(company.account_peppol_proxy_state, 'smp_registration')
with self._set_context({'participant_state': 'receiver'}):
self.env['account_edi_proxy_client.user']._cron_peppol_get_participant_status()
self.assertEqual(company.account_peppol_proxy_state, 'receiver')
def test_create_success_receiver_two_steps(self):
# it should be possible to first register as a sender in the wizard
# and then come back to settings and register as a receiver
# first step: use the peppol wizard to register only as a sender
company = self.env.company
wizard = self.env['peppol.registration'].create(self._get_participant_vals())
wizard.button_peppol_sender_registration()
wizard.verification_code = '123456'
wizard.button_check_peppol_verification_code()
self.assertEqual(company.account_peppol_proxy_state, 'sender')
# second step: open settings and register as a receiver
settings = self.env['res.config.settings'].create({})
settings.button_peppol_smp_registration()
self.assertEqual(company.account_peppol_proxy_state, 'smp_registration')
self.env['account_edi_proxy_client.user']._cron_peppol_get_participant_status()
self.assertEqual(company.account_peppol_proxy_state, 'receiver')
def test_create_reject_participant(self):
# the account_peppol_proxy_state should change to rejected
# if we reject the participant
company = self.env.company
wizard = self.env['peppol.registration'].create(self._get_participant_vals())
with self._set_context({'participant_state': 'rejected'}):
wizard.button_peppol_sender_registration()
company.account_peppol_proxy_state = 'smp_registration'
self.env['account_edi_proxy_client.user']._cron_peppol_get_participant_status()
self.assertEqual(company.account_peppol_proxy_state, 'rejected')
@mute_logger('odoo.sql_db')
def test_create_duplicate_participant(self):
# should not be possible to create a duplicate participant
wizard = self.env['peppol.registration'].create(self._get_participant_vals())
wizard.button_peppol_sender_registration()
with self.assertRaises(IntegrityError), self.cr.savepoint():
wizard.account_peppol_proxy_state = 'not_registered'
wizard.button_peppol_sender_registration()
def test_save_migration_key(self):
# migration key should be saved
wizard = self.env['peppol.registration']\
.create({
**self._get_participant_vals(),
'smp_registration': True,
'account_peppol_migration_key': 'helloo',
})
with self._set_context({'migrate_to': True}):
self.assertEqual(self.env.company.account_peppol_migration_key, 'helloo')
wizard.button_peppol_sender_registration()
wizard.verification_code = '123456'
wizard.button_check_peppol_verification_code()
self.assertEqual(self.env.company.account_peppol_proxy_state, 'smp_registration')
self.assertFalse(self.env.company.account_peppol_migration_key) # the key should be reset once we've used it
def test_migrate_away_participant(self):
# a participant should be able to request a migration key
wizard = self.env['peppol.registration'].create(self._get_participant_vals())
self.assertFalse(wizard.account_peppol_migration_key)
wizard.button_peppol_sender_registration()
wizard.account_peppol_proxy_state = 'receiver'
# migrating away is only possible in the settings
settings = self.env['res.config.settings'].create({})
settings.button_migrate_peppol_registration()
self.assertEqual(settings.company_id.account_peppol_proxy_state, 'receiver')
self.assertEqual(settings.account_peppol_migration_key, 'test_key')
def test_reset_participant(self):
# once a participant has migrated away, they should be reset
wizard = self.env['peppol.registration'].create(self._get_participant_vals())
wizard.button_peppol_sender_registration()
wizard.account_peppol_proxy_state = 'receiver'
settings = self.env['res.config.settings'].create({})
settings.button_migrate_peppol_registration()
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, [{
'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:
raise ValidationError('A UserError should be raised.')
|