Odoo18-Base/addons/account_peppol/tools/demo_utils.py
2025-01-06 10:57:38 +07:00

216 lines
8.4 KiB
Python

# 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,
}[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.invoice_edi_format
if endpoint and eas and edi_format:
self.with_company(company).peppol_verification_state = 'valid'
self.with_company(company).invoice_sending_method = 'peppol'
self._log_verification_state_update(company, old_value, 'valid')
else:
self.with_company(company).peppol_verification_state = 'not_valid'
def _mock_get_peppol_verification_state(func, self, *args, **kwargs):
(endpoint, eas, format) = args
return 'valid' if endpoint and eas and format else False
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.
"""
demo_mode = self.env.company._get_peppol_edi_mode() == 'demo'
if not demo_mode or modules.module.current_test:
return func(self, *args, **kwargs)
return _demo_behaviour[func.__name__](func, self, *args, **kwargs)