oca/oca-ocb-pos
2
0
Fork 0
mirror of https://github.com/bringout/oca-ocb-pos.git synced 2026-04-24 14:21:59 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

19.0 vanilla

Browse source
This commit is contained in:
Ernad Husremovic 2026-03-09 09:29:53 +01:00
parent 6e54c1af6c
commit 3ca647e428
1087 changed files with 132065 additions and 108499 deletions
Download patch file Download diff file Expand all files Collapse all files

61
odoo-bringout-oca-ocb-pos_adyen/pos_adyen/models/pos_payment_method.py
View file

@ -3,11 +3,8 @@
import json
import logging
import pprint
import random
import requests
import string
from urllib.parse import parse_qs
from werkzeug.exceptions import Forbidden
from odoo import fields, models, api, _
from odoo.exceptions import ValidationError, UserError, AccessDenied
@ -17,6 +14,7 @@ _logger = logging.getLogger(__name__)
UNPREDICTABLE_ADYEN_DATA = object() # sentinel
class PosPaymentMethod(models.Model):
_inherit = 'pos.payment.method'
@ -29,7 +27,19 @@ class PosPaymentMethod(models.Model):
adyen_test_mode = fields.Boolean(help='Run transactions in the test environment.', groups='base.group_erp_manager')
adyen_latest_response = fields.Char(copy=False, groups='base.group_erp_manager') # used to buffer the latest asynchronous notification from Adyen.
adyen_latest_diagnosis = fields.Char(copy=False, groups='base.group_erp_manager') # used to determine if the terminal is still connected.
adyen_event_url = fields.Char(
string="Event URL",
help="This URL needs to be pasted on Adyen's portal terminal settings.",
readonly=True,
store=False,
default=lambda self: f"{self.get_base_url()}/pos_adyen/notification",
)
@api.model
def _load_pos_data_fields(self, config):
params = super()._load_pos_data_fields(config)
params += ['adyen_terminal_identifier']
return params
@api.constrains('adyen_terminal_identifier')
def _check_adyen_terminal_identifier(self):
@ -42,13 +52,13 @@ class PosPaymentMethod(models.Model):
limit=1)
if existing_payment_method:
if existing_payment_method.company_id == payment_method.company_id:
raise ValidationError(_('Terminal %s is already used on payment method %s.')
% (payment_method.adyen_terminal_identifier, existing_payment_method.display_name))
raise ValidationError(_('Terminal %(terminal)s is already used on payment method %(payment_method)s.',
terminal=payment_method.adyen_terminal_identifier, payment_method=existing_payment_method.display_name))
else:
raise ValidationError(_('Terminal %s is already used in company %s on payment method %s.')
% (payment_method.adyen_terminal_identifier,
existing_payment_method.company_id.name,
existing_payment_method.display_name))
raise ValidationError(_('Terminal %(terminal)s is already used in company %(company)s on payment method %(payment_method)s.',
terminal=payment_method.adyen_terminal_identifier,
company=existing_payment_method.company_id.name,
payment_method=existing_payment_method.display_name))
def _get_adyen_endpoints(self):
return {
@ -56,25 +66,21 @@ class PosPaymentMethod(models.Model):
}
def _is_write_forbidden(self, fields):
whitelisted_fields = set(('adyen_latest_response', 'adyen_latest_diagnosis'))
return super(PosPaymentMethod, self)._is_write_forbidden(fields - whitelisted_fields)
return super(PosPaymentMethod, self)._is_write_forbidden(fields - {'adyen_latest_response'})
def get_latest_adyen_status(self):
self.ensure_one()
if not self.env.su and not self.user_has_groups('point_of_sale.group_pos_user'):
if not self.env.su and not self.env.user.has_group('point_of_sale.group_pos_user'):
raise AccessDenied()
latest_response = self.sudo().adyen_latest_response
latest_response = json.loads(latest_response) if latest_response else False
return {
'latest_response': latest_response,
}
return latest_response
def proxy_adyen_request(self, data, operation=False):
''' Necessary because Adyen's endpoints don't have CORS enabled '''
self.ensure_one()
if not self.env.su and not self.user_has_groups('point_of_sale.group_pos_user'):
if not self.env.su and not self.env.user.has_group('point_of_sale.group_pos_user'):
raise AccessDenied()
if not data:
raise UserError(_('Invalid Adyen request'))
@ -126,14 +132,18 @@ class PosPaymentMethod(models.Model):
if is_payment_request_with_acquirer_data:
parsed_sale_to_acquirer_data = parse_qs(data['SaleToPOIRequest']['PaymentRequest']['SaleData']['SaleToAcquirerData'])
is_payment_request_with_acquirer_data = len(parsed_sale_to_acquirer_data) <= 2
valid_acquirer_data = self._get_valid_acquirer_data()
is_payment_request_with_acquirer_data = len(parsed_sale_to_acquirer_data.keys()) <= len(valid_acquirer_data.keys())
if is_payment_request_with_acquirer_data:
for key, values in parsed_sale_to_acquirer_data.items():
if len(values) != 1:
is_payment_request_with_acquirer_data = False
break
value = values[0]
if not ((key == 'tenderOption' and value == 'AskGratuity') or (key == 'authorisationType' and value == 'PreAuth')):
valid_value = valid_acquirer_data.get(key)
if valid_value == UNPREDICTABLE_ADYEN_DATA:
continue
if value != valid_value:
is_payment_request_with_acquirer_data = False
break
@ -205,11 +215,18 @@ class PosPaymentMethod(models.Model):
return res
@api.model
def _get_hmac(self, sale_id, service_id, poiid, sale_transaction_id):
def _get_valid_acquirer_data(self):
return {
'tenderOption': 'AskGratuity',
'authorisationType': 'PreAuth'
}
@api.model
def _get_hmac(self, sale_id, service_id, poi_id, sale_transaction_id):
return hmac(
env=self.env(su=True),
scope='pos_adyen_payment',
message=(sale_id, service_id, poiid, sale_transaction_id),
message=(sale_id, service_id, poi_id, sale_transaction_id)
)
def _proxy_adyen_request_direct(self, data, operation):