# Part of Odoo. See LICENSE file for full copyright and licensing details. from base64 import b64encode from decorator import decorator import uuid from odoo import _, fields, modules from odoo.tools.misc import file_open DEMO_BILL_PATH = 'account_peppol/tools/demo_bill' DEMO_ENC_KEY = 'account_peppol/tools/enc_key' DEMO_PRIVATE_KEY = 'account_peppol/tools/private_key.pem' # ------------------------------------------------------------------------- # HELPERS # ------------------------------------------------------------------------- def get_demo_vendor_bill(user): return { 'direction': 'incoming', 'receiver': user.edi_identification, 'uuid': f'{user.company_id.id}_demo_vendor_bill', 'accounting_supplier_party': '0208:2718281828', 'state': 'done', 'filename': f'{user.company_id.id}_demo_vendor_bill', 'enc_key': file_open(DEMO_ENC_KEY, mode='rb').read(), 'document': file_open(DEMO_BILL_PATH, mode='rb').read(), } def _get_notification_message(proxy_state): if proxy_state == 'receiver': title = _("Registered to receive documents via Peppol (demo).") message = _("You can now receive demo vendor bills.") else: title = _("Registered as a sender (demo).") message = _("You can now send invoices in demo mode.") return title, message # ------------------------------------------------------------------------- # MOCKED FUNCTIONS # ------------------------------------------------------------------------- def _mock_call_peppol_proxy(func, self, *args, **kwargs): def _mock_get_all_documents(user, args, kwargs): if not user.env['account.move'].search_count([ ('peppol_message_uuid', '=', f'{user.company_id.id}_demo_vendor_bill') ]): return {'messages': [get_demo_vendor_bill(user)]} return {'messages': []} def _mock_get_document(user, args, kwargs): message_uuid = args[1]['message_uuids'][0] if message_uuid.endswith('_demo_vendor_bill'): return {message_uuid: get_demo_vendor_bill(user)} return {message_uuid: {'state': 'done'}} def _mock_send_document(user, args, kwargs): # Trigger the reception of vendor bills get_messages_cron = user.env['ir.cron'].sudo().env.ref( 'account_peppol.ir_cron_peppol_get_new_documents', raise_if_not_found=False, ) if get_messages_cron: get_messages_cron._trigger() return { 'messages': [{ 'message_uuid': 'demo_%s' % uuid.uuid4(), } for i in args[1]['documents']], } endpoint = args[0].split('/')[-1] return { 'ack': lambda _user, _args, _kwargs: {}, 'activate_participant': lambda _user, _args, _kwargs: {}, 'register_sender': lambda _user, _args, _kwargs: {}, 'register_receiver': lambda _user, _args, _kwargs: {}, 'register_sender_as_receiver': lambda _user, _args, _kwargs: {}, 'get_all_documents': _mock_get_all_documents, 'get_document': _mock_get_document, 'participant_status': lambda _user, _args, _kwargs: {'peppol_state': 'receiver'}, 'send_document': _mock_send_document, 'add_services': lambda _user, _args, _kwargs: {}, 'remove_services': lambda _user, _args, _kwargs: {}, }[endpoint](self, args, kwargs) def _mock_button_verify_partner_endpoint(func, self, *args, **kwargs): self.ensure_one() old_value = self.peppol_verification_state company = kwargs.get('company') or self.env.company endpoint, eas, edi_format = self.peppol_endpoint, self.peppol_eas, self._get_peppol_edi_format() state = _mock_get_peppol_verification_state(func, self, endpoint, eas, edi_format) self.with_company(company).peppol_verification_state = state self._log_verification_state_update(company, old_value, state) if state == 'valid': self.with_company(company).invoice_sending_method = 'peppol' def _mock_get_peppol_verification_state(func, self, *args, **kwargs): (endpoint, eas, xml_format) = args if not (eas and endpoint): return 'not_verified' if not xml_format: return 'not_valid' if xml_format not in self._get_peppol_formats(): return 'not_valid_format' return 'valid' def _mock_user_creation(func, self, *args, **kwargs): func(self, *args, **kwargs) self.account_peppol_proxy_state = 'receiver' if self.smp_registration else 'sender' content = b64encode(file_open(DEMO_PRIVATE_KEY, 'rb').read()) attachments = self.env['ir.attachment'].search([ ('res_model', '=', 'certificate.key'), ('res_field', '=', 'content'), ('company_id', '=', self.edi_user_id.company_id.id) ]) content_to_key_id = {attachment.datas: attachment.res_id for attachment in attachments} pkey_id = content_to_key_id.get(content) if not pkey_id: pkey_id = self.env['certificate.key'].create({ 'content': content, 'company_id': self.edi_user_id.company_id.id, }) self.edi_user_id.private_key_id = pkey_id return self._action_send_notification( *_get_notification_message(self.account_peppol_proxy_state) ) def _mock_receiver_registration(func, self, *args, **kwargs): self.account_peppol_proxy_state = 'receiver' return self.env['peppol.registration']._action_send_notification( *_get_notification_message(self.account_peppol_proxy_state) ) def _mock_check_verification_code(func, self, *args, **kwargs): self.button_peppol_sender_registration() self.verification_code = False return self.env['peppol.registration']._action_send_notification( *_get_notification_message(self.account_peppol_proxy_state) ) def _mock_deregister_participant(func, self, *args, **kwargs): # Set documents sent in demo to a state where they can be re-sent demo_moves = self.env['account.move'].search([ ('company_id', '=', self.company_id.id), ('peppol_message_uuid', '=like', 'demo_%'), ]) demo_moves.write({ 'peppol_message_uuid': None, 'peppol_move_state': None, }) demo_moves.message_main_attachment_id.unlink() demo_moves.ubl_cii_xml_id.unlink() log_message = _('The peppol status of the documents has been reset when switching from Demo to Live.') demo_moves._message_log_batch(bodies=dict((move.id, log_message) for move in demo_moves)) # also unlink the demo vendor bill self.env['account.move'].search([ ('company_id', '=', self.company_id.id), ('peppol_message_uuid', '=', f'{self.company_id.id}_demo_vendor_bill'), ]).unlink() mode_constraint = self.env['ir.config_parameter'].get_param('account_peppol.mode_constraint') if 'account_peppol_edi_user' in self._fields: self.account_peppol_edi_user.unlink() else: self.edi_user_id.unlink() self.account_peppol_proxy_state = 'not_registered' if 'account_peppol_edi_mode' in self._fields: self.account_peppol_edi_mode = mode_constraint def _mock_update_user_data(func, self, *args, **kwargs): pass def _mock_migrate_participant(func, self, *args, **kwargs): self.company_id.account_peppol_migration_key = 'demo_migration_key' def _mock_check_company_on_peppol(func, self, *args, **kwargs): pass _demo_behaviour = { '_call_peppol_proxy': _mock_call_peppol_proxy, 'button_account_peppol_check_partner_endpoint': _mock_button_verify_partner_endpoint, '_get_peppol_verification_state': _mock_get_peppol_verification_state, '_peppol_migrate_registration': _mock_migrate_participant, 'button_peppol_sender_registration': _mock_user_creation, 'button_deregister_peppol_participant': _mock_deregister_participant, 'button_update_peppol_user_data': _mock_update_user_data, 'button_peppol_smp_registration': _mock_receiver_registration, 'button_check_peppol_verification_code': _mock_check_verification_code, 'button_register_peppol_participant': _mock_user_creation, '_check_company_on_peppol': _mock_check_company_on_peppol, } # ------------------------------------------------------------------------- # DECORATORS # ------------------------------------------------------------------------- @decorator def handle_demo(func, self, *args, **kwargs): """ This decorator is used on methods that should be mocked in demo mode. First handle the decision: "Are we in demo mode?", and conditionally decide which function to execute. """ if self.env.company._get_peppol_edi_mode() == 'demo': return _demo_behaviour[func.__name__](func, self, *args, **kwargs) return func(self, *args, **kwargs)