OCA-git-bot
2 years ago
17 changed files with 968 additions and 334 deletions
-
1.pre-commit-config.yaml
-
17account_bank_statement_import_online_ponto/__manifest__.py
-
20account_bank_statement_import_online_ponto/data/ir_cron.xml
-
5account_bank_statement_import_online_ponto/models/__init__.py
-
329account_bank_statement_import_online_ponto/models/online_bank_statement_provider_ponto.py
-
77account_bank_statement_import_online_ponto/models/ponto_buffer.py
-
31account_bank_statement_import_online_ponto/models/ponto_buffer_line.py
-
206account_bank_statement_import_online_ponto/models/ponto_interface.py
-
3account_bank_statement_import_online_ponto/security/ir.model.access.csv
-
3account_bank_statement_import_online_ponto/tests/__init__.py
-
128account_bank_statement_import_online_ponto/tests/test_account_bank_statement_import_online_ponto.py
-
256account_bank_statement_import_online_ponto/tests/test_account_statement_import_online_ponto.py
-
150account_bank_statement_import_online_ponto/tests/test_ponto_interface.py
-
11account_bank_statement_import_online_ponto/views/ir_actions_act_window.xml
-
13account_bank_statement_import_online_ponto/views/ir_ui_menu.xml
-
17account_bank_statement_import_online_ponto/views/online_bank_statement_provider.xml
-
35account_bank_statement_import_online_ponto/views/ponto_buffer.xml
@ -1,15 +1,24 @@ |
|||
# Copyright 2020 Florent de Labarre |
|||
# Copyright 2020 Florent de Labarre. |
|||
# Copyright 2022 Therp BV <https://therp.nl>. |
|||
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). |
|||
{ |
|||
"name": "Online Bank Statements: MyPonto.com", |
|||
"version": "12.0.1.1.1", |
|||
"version": "12.0.1.2.0", |
|||
"category": "Account", |
|||
"website": "https://github.com/OCA/bank-statement-import", |
|||
"author": "Florent de Labarre, Odoo Community Association (OCA)", |
|||
"author": |
|||
"Florent de Labarre" |
|||
", Therp BV" |
|||
", Odoo Community Association (OCA)", |
|||
"license": "AGPL-3", |
|||
"installable": True, |
|||
"depends": ["account_bank_statement_import_online"], |
|||
"data": [ |
|||
"view/online_bank_statement_provider.xml" |
|||
"data/ir_cron.xml", |
|||
"security/ir.model.access.csv", |
|||
"views/online_bank_statement_provider.xml", |
|||
"views/ponto_buffer.xml", |
|||
"views/ir_actions_act_window.xml", |
|||
"views/ir_ui_menu.xml", |
|||
], |
|||
} |
|||
@ -0,0 +1,20 @@ |
|||
<?xml version="1.0" ?> |
|||
<!-- |
|||
Copyright 2022 Therp BV (https://therp.nl) |
|||
License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). |
|||
--> |
|||
<odoo noupdate="1"> |
|||
|
|||
<record model="ir.cron" id="ir_cron_purge_ponto_buffer"> |
|||
<field name="name">Remove old data from ponto buffers</field> |
|||
<field name="interval_number">1</field> |
|||
<field name="interval_type">days</field> |
|||
<field name="numbercall">-1</field> |
|||
<field name="state">code</field> |
|||
<field name="nextcall">2019-01-01 00:20:00</field> |
|||
<field name="doall" eval="False"/> |
|||
<field name="model_id" ref="account_bank_statement_import_online_ponto.model_online_bank_statement_provider"/> |
|||
<field name="code">model._ponto_buffer_purge()</field> |
|||
</record> |
|||
|
|||
</odoo> |
|||
@ -1,4 +1,5 @@ |
|||
# Copyright 2020 Florent de Labarre |
|||
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). |
|||
|
|||
from . import ponto_buffer |
|||
from . import ponto_buffer_line |
|||
from . import ponto_interface |
|||
from . import online_bank_statement_provider_ponto |
|||
@ -1,226 +1,171 @@ |
|||
# Copyright 2020 Florent de Labarre |
|||
# Copyright 2022 Therp BV <https://therp.nl>. |
|||
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). |
|||
|
|||
import requests |
|||
from datetime import date, datetime |
|||
from dateutil.relativedelta import relativedelta |
|||
import json |
|||
import base64 |
|||
import time |
|||
import re |
|||
import pytz |
|||
from datetime import datetime |
|||
|
|||
import logging |
|||
import re |
|||
|
|||
from odoo import api, fields, models, _ |
|||
from odoo.exceptions import UserError |
|||
from dateutil.relativedelta import relativedelta |
|||
from odoo.addons.base.models.res_bank import sanitize_account_number |
|||
|
|||
PONTO_ENDPOINT = 'https://api.myponto.com' |
|||
_logger = logging.getLogger(__name__) |
|||
|
|||
|
|||
class OnlineBankStatementProviderPonto(models.Model): |
|||
_inherit = 'online.bank.statement.provider' |
|||
|
|||
ponto_token = fields.Char(readonly=True) |
|||
ponto_token_expiration = fields.Datetime(readonly=True) |
|||
ponto_last_identifier = fields.Char(readonly=True) |
|||
_inherit = "online.bank.statement.provider" |
|||
|
|||
def ponto_reset_last_identifier(self): |
|||
self.write({'ponto_last_identifier': False}) |
|||
ponto_buffer_retain_days = fields.Integer( |
|||
string="Number of days to keep Ponto Buffers", |
|||
default=61, |
|||
help="By default buffers will be kept for 61 days.\n" |
|||
"Set this to 0 to keep buffers indefinitely.", |
|||
) |
|||
|
|||
@api.model |
|||
def _get_available_services(self): |
|||
return super()._get_available_services() + [ |
|||
('ponto', 'MyPonto.com'), |
|||
("ponto", "MyPonto.com"), |
|||
] |
|||
|
|||
@api.multi |
|||
def _pull(self, date_since, date_until): |
|||
"""Override pull to first retrieve data from Ponto.""" |
|||
if self.service == "ponto": |
|||
self._ponto_retrieve_data(date_since) |
|||
super()._pull(date_since, date_until) |
|||
|
|||
def _ponto_retrieve_data(self, date_since): |
|||
"""Fill buffer with data from Ponto. |
|||
|
|||
We will retrieve data from the latest transactions present in Ponto |
|||
backwards, until we find data that has an execution date before date_since. |
|||
""" |
|||
interface_model = self.env["ponto.interface"] |
|||
buffer_model = self.env["ponto.buffer"] |
|||
access_data = interface_model._login(self.username, self.password) |
|||
interface_model._set_access_account(access_data, self.account_number) |
|||
interface_model._ponto_synchronisation(access_data) |
|||
latest_identifier = False |
|||
transactions = interface_model._get_transactions( |
|||
access_data, |
|||
latest_identifier |
|||
) |
|||
while transactions: |
|||
buffer_model.sudo()._store_transactions(self, transactions) |
|||
latest_identifier = transactions[-1].get("id") |
|||
earliest_datetime = self._ponto_get_execution_datetime(transactions[-1]) |
|||
if earliest_datetime < date_since: |
|||
break |
|||
transactions = interface_model._get_transactions( |
|||
access_data, |
|||
latest_identifier |
|||
) |
|||
|
|||
def _obtain_statement_data(self, date_since, date_until): |
|||
self.ensure_one() |
|||
if self.service != 'ponto': |
|||
if self.service != "ponto": # pragma: no cover |
|||
return super()._obtain_statement_data( |
|||
date_since, |
|||
date_until, |
|||
) |
|||
return self._ponto_obtain_statement_data(date_since, date_until) |
|||
|
|||
######### |
|||
# ponto # |
|||
######### |
|||
|
|||
def _ponto_header_token(self): |
|||
self.ensure_one() |
|||
if self.username and self.password: |
|||
login = '%s:%s' % (self.username, self.password) |
|||
login = base64.b64encode(login.encode('UTF-8')).decode('UTF-8') |
|||
return {'Content-Type': 'application/x-www-form-urlencoded', |
|||
'Accept': 'application/json', |
|||
'Authorization': 'Basic %s' % login, } |
|||
raise UserError(_('Please fill login and key.')) |
|||
|
|||
def _ponto_header(self): |
|||
self.ensure_one() |
|||
if not self.ponto_token \ |
|||
or not self.ponto_token_expiration \ |
|||
or self.ponto_token_expiration <= fields.Datetime.now(): |
|||
|
|||
url = PONTO_ENDPOINT + '/oauth2/token' |
|||
response = requests.post(url, verify=False, |
|||
params={'grant_type': 'client_credentials'}, |
|||
headers=self._ponto_header_token()) |
|||
if response.status_code == 200: |
|||
data = json.loads(response.text) |
|||
access_token = data.get('access_token', False) |
|||
if not access_token: |
|||
raise UserError(_('Ponto : no token')) |
|||
else: |
|||
self.sudo().ponto_token = access_token |
|||
expiration_date = fields.Datetime.now() + relativedelta( |
|||
seconds=data.get('expires_in', False)) |
|||
self.sudo().ponto_token_expiration = expiration_date |
|||
else: |
|||
raise UserError(_('%s \n\n %s') % (response.status_code, response.text)) |
|||
return {'Accept': 'application/json', |
|||
'Authorization': 'Bearer %s' % self.ponto_token, } |
|||
|
|||
def _ponto_get_account_ids(self): |
|||
url = PONTO_ENDPOINT + '/accounts' |
|||
response = requests.get(url, verify=False, params={'limit': 100}, |
|||
headers=self._ponto_header()) |
|||
if response.status_code == 200: |
|||
data = json.loads(response.text) |
|||
res = {} |
|||
for account in data.get('data', []): |
|||
iban = sanitize_account_number( |
|||
account.get('attributes', {}).get('reference', '')) |
|||
res[iban] = account.get('id') |
|||
return res |
|||
raise UserError(_('%s \n\n %s') % (response.status_code, response.text)) |
|||
|
|||
def _ponto_synchronisation(self, account_id): |
|||
url = PONTO_ENDPOINT + '/synchronizations' |
|||
data = {'data': { |
|||
'type': 'synchronization', |
|||
'attributes': { |
|||
'resourceType': 'account', |
|||
'resourceId': account_id, |
|||
'subtype': 'accountTransactions' |
|||
} |
|||
}} |
|||
response = requests.post(url, verify=False, |
|||
headers=self._ponto_header(), |
|||
json=data) |
|||
if response.status_code in (200, 201, 400): |
|||
data = json.loads(response.text) |
|||
sync_id = data.get('attributes', {}).get('resourceId', False) |
|||
else: |
|||
raise UserError(_('Error during Create Synchronisation %s \n\n %s') % ( |
|||
response.status_code, response.text)) |
|||
|
|||
# Check synchronisation |
|||
if not sync_id: |
|||
return |
|||
url = PONTO_ENDPOINT + '/synchronizations/' + sync_id |
|||
number = 0 |
|||
while number == 100: |
|||
number += 1 |
|||
response = requests.get(url, verify=False, headers=self._ponto_header()) |
|||
if response.status_code == 200: |
|||
data = json.loads(response.text) |
|||
status = data.get('status', {}) |
|||
if status in ('success', 'error'): |
|||
return |
|||
time.sleep(4) |
|||
|
|||
def _ponto_get_transaction(self, account_id, date_since, date_until): |
|||
page_url = PONTO_ENDPOINT + '/accounts/' + account_id + '/transactions' |
|||
params = {'limit': 100} |
|||
page_next = True |
|||
last_identifier = self.ponto_last_identifier |
|||
if last_identifier: |
|||
params['before'] = last_identifier |
|||
page_next = False |
|||
transaction_lines = [] |
|||
latest_identifier = False |
|||
while page_url: |
|||
response = requests.get(page_url, verify=False, params=params, |
|||
headers=self._ponto_header()) |
|||
if response.status_code == 200: |
|||
if params.get('before'): |
|||
params.pop('before') |
|||
data = json.loads(response.text) |
|||
links = data.get('links', {}) |
|||
if page_next: |
|||
page_url = links.get('next', False) |
|||
else: |
|||
page_url = links.get('prev', False) |
|||
transactions = data.get('data', []) |
|||
if transactions: |
|||
current_transactions = [] |
|||
for transaction in transactions: |
|||
date = self._ponto_date_from_string( |
|||
transaction.get('attributes', {}).get('executionDate')) |
|||
if date_since <= date < date_until: |
|||
current_transactions.append(transaction) |
|||
|
|||
if current_transactions: |
|||
if not page_next or (page_next and not latest_identifier): |
|||
latest_identifier = current_transactions[0].get('id') |
|||
transaction_lines.extend(current_transactions) |
|||
else: |
|||
raise UserError( |
|||
_('Error during get transaction.\n\n%s \n\n %s') % ( |
|||
response.status_code, response.text)) |
|||
if latest_identifier: |
|||
self.ponto_last_identifier = latest_identifier |
|||
return transaction_lines |
|||
|
|||
def _ponto_date_from_string(self, date_str): |
|||
"""Dates in Ponto are expressed in UTC, so we need to convert them |
|||
to supplied tz for proper classification. |
|||
""" |
|||
dt = datetime.strptime(date_str, '%Y-%m-%dT%H:%M:%S.%fZ') |
|||
dt = dt.replace(tzinfo=pytz.utc).astimezone( |
|||
pytz.timezone(self.tz or 'utc') |
|||
) |
|||
return dt.replace(tzinfo=None) |
|||
|
|||
def _ponto_obtain_statement_data(self, date_since, date_until): |
|||
"""Translate information from Ponto to Odoo bank statement lines.""" |
|||
self.ensure_one() |
|||
account_ids = self._ponto_get_account_ids() |
|||
journal = self.journal_id |
|||
iban = self.account_number |
|||
account_id = account_ids.get(iban) |
|||
if not account_id: |
|||
raise UserError( |
|||
_('Ponto : wrong configuration, unknow account %s') |
|||
% journal.bank_account_id.acc_number) |
|||
self._ponto_synchronisation(account_id) |
|||
transaction_lines = self._ponto_get_transaction( |
|||
account_id, date_since, date_until) |
|||
_logger.debug( |
|||
_("Ponto obtain statement data for journal %s from %s to %s"), |
|||
self.journal_id.name, |
|||
date_since, |
|||
date_until |
|||
) |
|||
line_model = self.env["ponto.buffer.line"] |
|||
lines = line_model.sudo().search( |
|||
[ |
|||
("buffer_id.provider_id", "=", self.id), |
|||
("effective_date_time", ">=", date_since), |
|||
("effective_date_time", "<=", date_until), |
|||
] |
|||
) |
|||
new_transactions = [] |
|||
sequence = 0 |
|||
for transaction in transaction_lines: |
|||
for transaction in lines: |
|||
sequence += 1 |
|||
attributes = transaction.get('attributes', {}) |
|||
ref_list = [attributes.get(x) for x in { |
|||
vals_line = self._ponto_get_transaction_vals( |
|||
json.loads(transaction.transaction_data), |
|||
sequence |
|||
) |
|||
new_transactions.append(vals_line) |
|||
return new_transactions, {} |
|||
|
|||
def _ponto_get_transaction_vals(self, transaction, sequence): |
|||
"""Translate information from Ponto to statement line vals.""" |
|||
attributes = transaction.get("attributes", {}) |
|||
ref_list = [ |
|||
attributes.get(x) |
|||
for x in { |
|||
"description", |
|||
"counterpartName", |
|||
"counterpartReference", |
|||
} if attributes.get(x)] |
|||
ref = " ".join(ref_list) |
|||
date = self._ponto_date_from_string(attributes.get('executionDate')) |
|||
vals_line = { |
|||
'sequence': sequence, |
|||
'date': date, |
|||
'ref': re.sub(' +', ' ', ref) or '/', |
|||
'name': attributes.get('remittanceInformation') or ref, |
|||
'unique_import_id': transaction['id'], |
|||
'amount': attributes['amount'], |
|||
} |
|||
if attributes.get("counterpartReference"): |
|||
vals_line["account_number"] = attributes["counterpartReference"] |
|||
if attributes.get("counterpartName"): |
|||
vals_line["partner_name"] = attributes["counterpartName"] |
|||
new_transactions.append(vals_line) |
|||
if new_transactions: |
|||
return new_transactions, {} |
|||
return |
|||
if attributes.get(x) |
|||
] |
|||
ref = " ".join(ref_list) |
|||
date = self._ponto_get_execution_datetime(transaction) |
|||
vals_line = { |
|||
"sequence": sequence, |
|||
"date": date, |
|||
"ref": re.sub(" +", " ", ref) or "/", |
|||
"name": attributes.get("remittanceInformation") or ref, |
|||
"unique_import_id": transaction["id"], |
|||
"amount": attributes["amount"], |
|||
} |
|||
if attributes.get("counterpartReference"): |
|||
vals_line["account_number"] = attributes["counterpartReference"] |
|||
if attributes.get("counterpartName"): |
|||
vals_line["partner_name"] = attributes["counterpartName"] |
|||
return vals_line |
|||
|
|||
def _ponto_get_execution_datetime(self, transaction): |
|||
"""Get execution datetime for a transaction. |
|||
|
|||
Odoo often names variables containing date and time just xxx_date or |
|||
date_xxx. We try to avoid this misleading naming by using datetime as |
|||
much for variables and fields of type datetime. |
|||
""" |
|||
attributes = transaction.get("attributes", {}) |
|||
return self._ponto_datetime_from_string(attributes.get("executionDate")) |
|||
|
|||
def _ponto_datetime_from_string(self, date_str): |
|||
"""Dates in Ponto are expressed in UTC, so we need to convert them |
|||
to supplied tz for proper classification. |
|||
""" |
|||
dt = datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S.%fZ") |
|||
dt = dt.replace(tzinfo=pytz.utc).astimezone(pytz.timezone(self.tz or "utc")) |
|||
return dt.replace(tzinfo=None) |
|||
|
|||
def _ponto_buffer_purge(self): |
|||
"""Remove buffers from Ponto no longer needed to import statements.""" |
|||
_logger.info("Scheduled purge of old ponto buffers...") |
|||
today = date.today() |
|||
buffer_model = self.env["ponto.buffer"] |
|||
providers = self.search([ |
|||
("active", "=", True), |
|||
]) |
|||
for provider in providers: |
|||
if provider.service != "ponto": |
|||
continue |
|||
if not provider.ponto_buffer_retain_days: |
|||
continue |
|||
cutoff_date = today - relativedelta(days=provider.ponto_buffer_retain_days) |
|||
old_buffers = buffer_model.search( |
|||
[ |
|||
("provider_id", "=", provider.id), |
|||
("effective_date", "<", cutoff_date), |
|||
] |
|||
) |
|||
old_buffers.unlink() |
|||
_logger.info("Scheduled purge of old ponto buffers complete.") |
|||
@ -0,0 +1,77 @@ |
|||
# Copyright 2022 Therp BV <https://therp.nl>. |
|||
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). |
|||
"""Define model to hold data retrieved from Ponto.""" |
|||
import json |
|||
import logging |
|||
|
|||
from odoo import _, fields, models |
|||
|
|||
_logger = logging.getLogger(__name__) |
|||
|
|||
|
|||
class PontoBuffer(models.Model): |
|||
"""Define model to hold data retrieved from Ponto.""" |
|||
_name = "ponto.buffer" |
|||
_description = "Group transactions retrieved from Ponto." |
|||
_order = "effective_date desc" |
|||
_rec_name = "effective_date" |
|||
|
|||
provider_id = fields.Many2one( |
|||
comodel_name="online.bank.statement.provider", |
|||
required=True, |
|||
readonly=True, |
|||
) |
|||
effective_date = fields.Date(readonly=True, required=True) |
|||
buffer_line_ids = fields.One2many( |
|||
comodel_name="ponto.buffer.line", |
|||
inverse_name="buffer_id", |
|||
readonly=True, |
|||
) |
|||
|
|||
def _store_transactions(self, provider, transactions): |
|||
"""Store transactions retrieved from Ponto in buffer, preventing duplicates.""" |
|||
# Start by sorting all transactions per date. |
|||
transactions_per_date = {} |
|||
for transaction in transactions: |
|||
effective_date_time = provider._ponto_get_execution_datetime(transaction) |
|||
transaction["effective_date_time"] = effective_date_time.isoformat() |
|||
key = effective_date_time.isoformat()[0:10] |
|||
if key not in transactions_per_date: |
|||
transactions_per_date[key] = [] # Initialize transaction array. |
|||
transactions_per_date[key].append(transaction) |
|||
# Now store the transactions, but not when already present. |
|||
for key, date_transactions in transactions_per_date.items(): |
|||
_logger.debug( |
|||
_("For date %s we retrieved %d transactions"), |
|||
key, |
|||
len(date_transactions) |
|||
) |
|||
ponto_buffer = self.search( |
|||
[ |
|||
("provider_id", "=", provider.id), |
|||
("effective_date", "=", key), |
|||
], |
|||
limit=1 |
|||
) or self.create( |
|||
{ |
|||
"provider_id": provider.id, |
|||
"effective_date": key, |
|||
} |
|||
) |
|||
already_present = ponto_buffer.buffer_line_ids.mapped("ponto_id") |
|||
new_lines = [ |
|||
( |
|||
0, |
|||
0, |
|||
{ |
|||
"buffer_id": ponto_buffer.id, |
|||
"ponto_id": t.get("id"), |
|||
"effective_date_time": t.get("effective_date_time"), |
|||
"transaction_data": json.dumps(t), |
|||
}, |
|||
) |
|||
for t in date_transactions |
|||
if t.get("id") not in already_present |
|||
] |
|||
if new_lines: |
|||
ponto_buffer.write({"buffer_line_ids": new_lines}) |
|||
@ -0,0 +1,31 @@ |
|||
# Copyright 2022 Therp BV <https://therp.nl>. |
|||
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). |
|||
"""Define model to hold transactions retrieved from Ponto.""" |
|||
from odoo import fields, models |
|||
|
|||
|
|||
class PontoBuffer(models.Model): |
|||
"""Define model to hold transactions retrieved from Ponto.""" |
|||
_name = "ponto.buffer.line" |
|||
_description = "Hold transactions retrieved from Ponto." |
|||
_order = "effective_date_time desc" |
|||
_rec_name = "effective_date_time" |
|||
|
|||
buffer_id = fields.Many2one( |
|||
comodel_name="ponto.buffer", |
|||
required=True, |
|||
readonly=True, |
|||
ondelete="cascade", |
|||
) |
|||
ponto_id = fields.Char( |
|||
required=True, |
|||
readonly=True, |
|||
) |
|||
effective_date_time = fields.Datetime( |
|||
required=True, |
|||
readonly=True, |
|||
) |
|||
transaction_data = fields.Char( |
|||
required=True, |
|||
readonly=True, |
|||
) |
|||
@ -0,0 +1,206 @@ |
|||
# Copyright 2020 Florent de Labarre |
|||
# Copyright 2022 Therp BV <https://therp.nl>. |
|||
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). |
|||
|
|||
import base64 |
|||
import json |
|||
import logging |
|||
import time |
|||
|
|||
import requests |
|||
from dateutil.relativedelta import relativedelta |
|||
|
|||
from odoo import fields, models, _ |
|||
from odoo.exceptions import UserError |
|||
from odoo.addons.base.models.res_bank import sanitize_account_number |
|||
|
|||
_logger = logging.getLogger(__name__) |
|||
|
|||
PONTO_ENDPOINT = "https://api.myponto.com" |
|||
|
|||
|
|||
class PontoInterface(models.AbstractModel): |
|||
_name = "ponto.interface" |
|||
_description = "Interface to all interactions with Ponto API" |
|||
|
|||
def _login(self, username, password): |
|||
"""Ponto login returns an access dictionary for further requests.""" |
|||
url = PONTO_ENDPOINT + "/oauth2/token" |
|||
if not(username and password): |
|||
raise UserError(_("Please fill login and key.")) |
|||
login = "%s:%s" % (username, password) |
|||
login = base64.b64encode(login.encode("UTF-8")).decode("UTF-8") |
|||
login_headers = { |
|||
"Content-Type": "application/x-www-form-urlencoded", |
|||
"Accept": "application/json", |
|||
"Authorization": "Basic %s" % login, |
|||
} |
|||
response = requests.post( |
|||
url, |
|||
params={"grant_type": "client_credentials"}, |
|||
headers=login_headers, |
|||
) |
|||
data = self._get_response_data(response) |
|||
access_token = data.get("access_token", False) |
|||
if not access_token: |
|||
raise UserError(_("Ponto : no token")) |
|||
token_expiration = fields.Datetime.now() + relativedelta( |
|||
seconds=data.get("expires_in", False) |
|||
) |
|||
return { |
|||
"username": username, |
|||
"password": password, |
|||
"access_token": access_token, |
|||
"token_expiration": token_expiration, |
|||
} |
|||
|
|||
def _get_request_headers(self, access_data): |
|||
"""Get headers with authorization for further ponto requests.""" |
|||
if access_data["token_expiration"] <= fields.Datetime.now(): |
|||
updated_data = self._login(access_data["username"], access_data["password"]) |
|||
access_data.update(updated_data) |
|||
return { |
|||
"Accept": "application/json", |
|||
"Authorization": "Bearer %s" % access_data["access_token"], |
|||
} |
|||
|
|||
def _set_access_account(self, access_data, account_number): |
|||
"""Set ponto account for bank account in access_data.""" |
|||
url = PONTO_ENDPOINT + "/accounts" |
|||
response = requests.get( |
|||
url, params={"limit": 100}, headers=self._get_request_headers(access_data) |
|||
) |
|||
data = self._get_response_data(response) |
|||
for ponto_account in data.get("data", []): |
|||
ponto_iban = sanitize_account_number( |
|||
ponto_account.get("attributes", {}).get("reference", "") |
|||
) |
|||
if ponto_iban == account_number: |
|||
access_data["ponto_account"] = ponto_account.get("id") |
|||
return |
|||
# If we get here, we did not find Ponto account for bank account. |
|||
raise UserError( |
|||
_("Ponto : wrong configuration, account %s not found in %s") |
|||
% (account_number, data) |
|||
) |
|||
|
|||
def _ponto_synchronisation(self, access_data): |
|||
"""Make sure Ponto has retrieved latest data from financial institution.""" |
|||
url = PONTO_ENDPOINT + "/synchronizations" |
|||
# TODO: According to spec we should provide an IP number in the data. |
|||
# See: https://documentation.myponto.com/1/api/curl#create-synchronization |
|||
payload = { |
|||
"data": { |
|||
"type": "synchronization", |
|||
"attributes": { |
|||
"resourceType": "account", |
|||
"resourceId": access_data["ponto_account"], |
|||
"subtype": "accountTransactions", |
|||
}, |
|||
} |
|||
} |
|||
response = requests.post( |
|||
url, headers=self._get_request_headers(access_data), json=payload |
|||
) |
|||
if response.status_code == 400: |
|||
# Probably synchronization recently done already. |
|||
_logger.debug( |
|||
_("Synchronization request rejected: %s"), |
|||
response.text |
|||
) |
|||
return |
|||
data = self._get_response_data(response) |
|||
# The resourceId in data["attributes"] is the account, |
|||
# we need the synchronization. |
|||
sync_id = data.get("data", {}).get("id", False) |
|||
if not sync_id: |
|||
raise UserError( |
|||
_("Ponto : no synchronization id in data %s") % data |
|||
) |
|||
# Poll synchronization during 400 seconds for completion. |
|||
url = PONTO_ENDPOINT + "/synchronizations/" + sync_id |
|||
number = 0 |
|||
while number < 10: |
|||
number += 1 |
|||
if self._synchronization_done(access_data, url): |
|||
break |
|||
time.sleep(40) |
|||
|
|||
def _synchronization_done(self, access_data, url): |
|||
"""Check wether requested synchronization done.""" |
|||
response = requests.get(url, headers=self._get_request_headers(access_data)) |
|||
if response.status_code != 200: |
|||
return False |
|||
data = json.loads(response.text) |
|||
status = data.get("status", {}) |
|||
if status not in ("success", "error"): |
|||
return False |
|||
if status == "error": |
|||
_logger.debug( |
|||
_("Synchronization was succesfully completed") |
|||
) |
|||
else: |
|||
_logger.debug( |
|||
_("Synchronization had an error: %s"), |
|||
response.text |
|||
) |
|||
return True |
|||
|
|||
def _get_transactions(self, access_data, last_identifier): |
|||
"""Get transactions from ponto, using last_identifier as pointer. |
|||
|
|||
Note that Ponto has the transactions in descending order. The first |
|||
transaction, retrieved by not passing an identifier, is the latest |
|||
present in Ponto. If you read transactions 'after' a certain identifier |
|||
(Ponto id), you will get transactions with an earlier date. |
|||
""" |
|||
url = ( |
|||
PONTO_ENDPOINT |
|||
+ "/accounts/" |
|||
+ access_data["ponto_account"] |
|||
+ "/transactions" |
|||
) |
|||
params = {"limit": 100} |
|||
if last_identifier: |
|||
params["after"] = last_identifier |
|||
data = self._get_request(access_data, url, params) |
|||
transactions = self._get_transactions_from_data(data) |
|||
return transactions |
|||
|
|||
def _get_transactions_from_data(self, data): |
|||
"""Get all transactions that are in the ponto response data.""" |
|||
transactions = data.get("data", []) |
|||
if not transactions: |
|||
_logger.debug( |
|||
_("No transactions where found in data %s"), |
|||
data, |
|||
) |
|||
else: |
|||
_logger.debug( |
|||
_("%d transactions present in response data"), |
|||
len(transactions), |
|||
) |
|||
return transactions |
|||
|
|||
def _get_request(self, access_data, url, params): |
|||
"""Interact with Ponto to get next page of data.""" |
|||
headers = self._get_request_headers(access_data) |
|||
_logger.debug( |
|||
_("Get request to %s, with headers %s and params %s"), |
|||
url, |
|||
params, |
|||
headers |
|||
) |
|||
response = requests.get( |
|||
url, params=params, headers=headers |
|||
) |
|||
return self._get_response_data(response) |
|||
|
|||
def _get_response_data(self, response): |
|||
"""Get response data for GET or POST request.""" |
|||
if response.status_code not in (200, 201): |
|||
raise UserError( |
|||
_("Server returned status code %s: %s") |
|||
% (response.status_code, response.text) |
|||
) |
|||
return json.loads(response.text) |
|||
@ -0,0 +1,3 @@ |
|||
id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink |
|||
access_ponto_buffer,access_ponto_buffer,model_ponto_buffer,base.group_user,1,0,0,0 |
|||
access_ponto_buffer_line,access_ponto_buffer_line,model_ponto_buffer_line,base.group_user,1,0,0,0 |
|||
@ -1,3 +1,4 @@ |
|||
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). |
|||
|
|||
from . import test_account_bank_statement_import_online_ponto |
|||
from . import test_ponto_interface |
|||
from . import test_account_statement_import_online_ponto |
|||
@ -1,128 +0,0 @@ |
|||
# Copyright 2020 Florent de Labarre |
|||
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). |
|||
|
|||
from datetime import datetime |
|||
from unittest import mock |
|||
|
|||
from odoo import fields |
|||
from odoo.tests import common |
|||
|
|||
_module_ns = 'odoo.addons.account_bank_statement_import_online_ponto' |
|||
_provider_class = ( |
|||
_module_ns |
|||
+ '.models.online_bank_statement_provider_ponto' |
|||
+ '.OnlineBankStatementProviderPonto' |
|||
) |
|||
|
|||
|
|||
class TestAccountBankAccountStatementImportOnlineQonto( |
|||
common.TransactionCase |
|||
): |
|||
|
|||
def setUp(self): |
|||
super().setUp() |
|||
|
|||
self.now = fields.Datetime.now() |
|||
self.currency_eur = self.env.ref('base.EUR') |
|||
self.currency_usd = self.env.ref('base.USD') |
|||
self.AccountJournal = self.env['account.journal'] |
|||
self.ResPartnerBank = self.env['res.partner.bank'] |
|||
self.OnlineBankStatementProvider = self.env[ |
|||
'online.bank.statement.provider' |
|||
] |
|||
self.AccountBankStatement = self.env['account.bank.statement'] |
|||
self.AccountBankStatementLine = self.env['account.bank.statement.line'] |
|||
|
|||
self.bank_account = self.ResPartnerBank.create( |
|||
{'acc_number': 'FR0214508000302245362775K46', |
|||
'partner_id': self.env.user.company_id.partner_id.id}) |
|||
self.journal = self.AccountJournal.create({ |
|||
'name': 'Bank', |
|||
'type': 'bank', |
|||
'code': 'BANK', |
|||
'currency_id': self.currency_eur.id, |
|||
'bank_statements_source': 'online', |
|||
'online_bank_statement_provider': 'ponto', |
|||
'bank_account_id': self.bank_account.id, |
|||
}) |
|||
self.provider = self.journal.online_bank_statement_provider_id |
|||
|
|||
self.mock_header = lambda: mock.patch( |
|||
_provider_class + '._ponto_header', |
|||
return_value={'Accept': 'application/json', |
|||
'Authorization': 'Bearer --TOKEN--'}, |
|||
) |
|||
|
|||
self.mock_account_ids = lambda: mock.patch( |
|||
_provider_class + '._ponto_get_account_ids', |
|||
return_value={'FR0214508000302245362775K46': 'id'}, |
|||
) |
|||
self.mock_synchronisation = lambda: mock.patch( |
|||
_provider_class + '._ponto_synchronisation', |
|||
return_value=None, |
|||
) |
|||
|
|||
self.mock_transaction = lambda: mock.patch( |
|||
_provider_class + '._ponto_get_transaction', |
|||
return_value=[{ |
|||
'type': 'transaction', |
|||
'relationships': {'account': { |
|||
'links': { |
|||
'related': 'https://api.myponto.com/accounts/'}, |
|||
'data': {'type': 'account', |
|||
'id': 'fd3d5b1d-fca9-4310-a5c8-76f2a9dc7c75'}}}, |
|||
'id': '701ab965-21c4-46ca-b157-306c0646e0e2', |
|||
'attributes': {'valueDate': '2019-11-18T00:00:00.000Z', |
|||
'remittanceInformationType': 'unstructured', |
|||
'remittanceInformation': 'Minima vitae totam!', |
|||
'executionDate': '2019-11-20T00:00:00.000Z', |
|||
'description': 'Wire transfer', |
|||
'currency': 'EUR', |
|||
'counterpartReference': 'BE26089479973169', |
|||
'counterpartName': 'Osinski Group', |
|||
'amount': 6.08}}, |
|||
{'type': 'transaction', |
|||
'relationships': { |
|||
'account': {'links': { |
|||
'related': 'https://api.myponto.com/accounts/'}, |
|||
'data': { |
|||
'type': 'account', |
|||
'id': 'fd3d5b1d-fca9-4310-a5c8-76f2a9dc7c75'}}}, |
|||
'id': '9ac50483-16dc-4a82-aa60-df56077405cd', |
|||
'attributes': { |
|||
'valueDate': '2019-11-04T00:00:00.000Z', |
|||
'remittanceInformationType': 'unstructured', |
|||
'remittanceInformation': 'Quia voluptatem blanditiis.', |
|||
'executionDate': '2019-11-06T00:00:00.000Z', |
|||
'description': 'Wire transfer', |
|||
'currency': 'EUR', |
|||
'counterpartReference': 'BE97201830401438', |
|||
'counterpartName': 'Stokes-Miller', |
|||
'amount': 5.48}}, |
|||
{'type': 'transaction', 'relationships': {'account': {'links': { |
|||
'related': 'https://api.myponto.com/accounts/'}, |
|||
'data': { |
|||
'type': 'account', |
|||
'id': 'fd3d5b1d-fca9-4310-a5c8-76f2a9dc7c75'}}}, |
|||
'id': 'b21a6c65-1c52-4ba6-8cbc-127d2b2d85ff', |
|||
'attributes': { |
|||
'valueDate': '2019-11-04T00:00:00.000Z', |
|||
'remittanceInformationType': 'unstructured', |
|||
'remittanceInformation': 'Laboriosam repelo?', |
|||
'executionDate': '2019-11-04T00:00:00.000Z', |
|||
'description': 'Wire transfer', 'currency': 'EUR', |
|||
'counterpartReference': 'BE10325927501996', |
|||
'counterpartName': 'Strosin-Veum', 'amount': 5.83}}], |
|||
) |
|||
|
|||
def test_ponto(self): |
|||
with self.mock_transaction(), \ |
|||
self.mock_header(),\ |
|||
self.mock_synchronisation(), \ |
|||
self.mock_account_ids(): |
|||
lines, statement_values = self.provider._obtain_statement_data( |
|||
datetime(2019, 11, 3), |
|||
datetime(2019, 11, 17), |
|||
) |
|||
|
|||
self.assertEqual(len(lines), 3) |
|||
@ -0,0 +1,256 @@ |
|||
# Copyright 2020 Florent de Labarre |
|||
# Copyright 2022 Therp BV <https://therp.nl>. |
|||
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). |
|||
|
|||
from datetime import date, datetime |
|||
from unittest import mock |
|||
|
|||
from odoo import fields |
|||
from odoo.tests import Form, common |
|||
|
|||
_module_ns = "odoo.addons.account_bank_statement_import_online_ponto" |
|||
_interface_class = ( |
|||
_module_ns |
|||
+ ".models.ponto_interface" |
|||
+ ".PontoInterface" |
|||
) |
|||
|
|||
THREE_TRANSACTIONS = [ |
|||
{ |
|||
"type": "transaction", |
|||
"relationships": { |
|||
"account": { |
|||
"links": {"related": "https://api.myponto.com/accounts/"}, |
|||
"data": { |
|||
"type": "account", |
|||
"id": "fd3d5b1d-fca9-4310-a5c8-76f2a9dc7c75", |
|||
}, |
|||
} |
|||
}, |
|||
"id": "701ab965-21c4-46ca-b157-306c0646e0e2", |
|||
"attributes": { |
|||
"valueDate": "2019-11-18T00:00:00.000Z", |
|||
"remittanceInformationType": "unstructured", |
|||
"remittanceInformation": "Minima vitae totam!", |
|||
"executionDate": "2019-11-20T00:00:00.000Z", |
|||
"description": "Wire transfer", |
|||
"currency": "EUR", |
|||
"counterpartReference": "BE26089479973169", |
|||
"counterpartName": "Osinski Group", |
|||
"amount": 6.08, |
|||
}, |
|||
}, |
|||
{ |
|||
"type": "transaction", |
|||
"relationships": { |
|||
"account": { |
|||
"links": {"related": "https://api.myponto.com/accounts/"}, |
|||
"data": { |
|||
"type": "account", |
|||
"id": "fd3d5b1d-fca9-4310-a5c8-76f2a9dc7c75", |
|||
}, |
|||
} |
|||
}, |
|||
"id": "9ac50483-16dc-4a82-aa60-df56077405cd", |
|||
"attributes": { |
|||
"valueDate": "2019-11-04T00:00:00.000Z", |
|||
"remittanceInformationType": "unstructured", |
|||
"remittanceInformation": "Quia voluptatem blanditiis.", |
|||
"executionDate": "2019-11-06T00:00:00.000Z", |
|||
"description": "Wire transfer", |
|||
"currency": "EUR", |
|||
"counterpartReference": "BE97201830401438", |
|||
"counterpartName": "Stokes-Miller", |
|||
"amount": 5.48, |
|||
}, |
|||
}, |
|||
{ |
|||
"type": "transaction", |
|||
"relationships": { |
|||
"account": { |
|||
"links": {"related": "https://api.myponto.com/accounts/"}, |
|||
"data": { |
|||
"type": "account", |
|||
"id": "fd3d5b1d-fca9-4310-a5c8-76f2a9dc7c75", |
|||
}, |
|||
} |
|||
}, |
|||
"id": "b21a6c65-1c52-4ba6-8cbc-127d2b2d85ff", |
|||
"attributes": { |
|||
"valueDate": "2019-11-04T00:00:00.000Z", |
|||
"remittanceInformationType": "unstructured", |
|||
"remittanceInformation": "Laboriosam repelo?", |
|||
"executionDate": "2019-11-04T00:00:00.000Z", |
|||
"description": "Wire transfer", |
|||
"currency": "EUR", |
|||
"counterpartReference": "BE10325927501996", |
|||
"counterpartName": "Strosin-Veum", |
|||
"amount": 5.83, |
|||
}, |
|||
}, |
|||
] |
|||
|
|||
EMPTY_TRANSACTIONS = [] |
|||
|
|||
|
|||
class TestBankAccountStatementImportOnlinePonto(common.TransactionCase): |
|||
post_install = True |
|||
|
|||
def setUp(self): |
|||
super().setUp() |
|||
|
|||
self.now = fields.Datetime.now() |
|||
self.currency_eur = self.env.ref("base.EUR") |
|||
self.currency_usd = self.env.ref("base.USD") |
|||
self.AccountJournal = self.env["account.journal"] |
|||
self.ResPartnerBank = self.env["res.partner.bank"] |
|||
self.OnlineBankStatementProvider = self.env["online.bank.statement.provider"] |
|||
self.AccountAccount = self.env["account.account"] |
|||
self.AccountBankStatement = self.env["account.bank.statement"] |
|||
self.AccountBankStatementLine = self.env["account.bank.statement.line"] |
|||
self.AccountStatementPull = self.env["online.bank.statement.pull.wizard"] |
|||
|
|||
self.bank_account = self.ResPartnerBank.create( |
|||
{ |
|||
"acc_number": "FR0214508000302245362775K46", |
|||
"partner_id": self.env.user.company_id.partner_id.id, |
|||
} |
|||
) |
|||
self.journal = self.AccountJournal.create( |
|||
{ |
|||
"name": "Bank", |
|||
"type": "bank", |
|||
"code": "BANK", |
|||
"currency_id": self.currency_eur.id, |
|||
"bank_statements_source": "online", |
|||
"online_bank_statement_provider": "ponto", |
|||
"bank_account_id": self.bank_account.id, |
|||
} |
|||
) |
|||
self.receivable_account = self.AccountAccount.create( |
|||
{ |
|||
"code": "1325", |
|||
"name": "Test receivable account", |
|||
"user_type_id": self.env.ref("account.data_account_type_payable").id, |
|||
"reconcile": True, |
|||
} |
|||
) |
|||
self.provider = self.journal.online_bank_statement_provider_id |
|||
|
|||
self.mock_login = lambda: mock.patch( |
|||
_interface_class + "._login", |
|||
return_value={ |
|||
"username": "test_user", |
|||
"password": "very_secret", |
|||
"access_token": "abcd1234", |
|||
"token_expiration": datetime(2099, 12, 31, 23, 59, 59), |
|||
}, |
|||
) |
|||
self.mock_set_access_account = lambda: mock.patch( |
|||
_interface_class + "._set_access_account", |
|||
return_value=None, |
|||
) |
|||
self.mock_synchronisation = lambda: mock.patch( |
|||
_interface_class + "._ponto_synchronisation", |
|||
return_value=None, |
|||
) |
|||
# return list of transactions on first call, empty list on second call. |
|||
self.mock_get_transactions = lambda: mock.patch( |
|||
_interface_class + "._get_transactions", |
|||
side_effect=[THREE_TRANSACTIONS, EMPTY_TRANSACTIONS, ] |
|||
) |
|||
|
|||
def test_balance_start(self): |
|||
st_form = Form(self.AccountBankStatement) |
|||
st_form.journal_id = self.journal |
|||
st_form.date = date(2019, 11, 1) |
|||
st_form.balance_end_real = 100 |
|||
with st_form.line_ids.new() as line_form: |
|||
line_form.name = "test move" |
|||
line_form.amount = 100 |
|||
initial_statement = st_form.save() |
|||
initial_statement.line_ids[0].account_id = self.receivable_account |
|||
initial_statement.button_confirm_bank() |
|||
with self.mock_login(), \ |
|||
self.mock_synchronisation(), \ |
|||
self.mock_set_access_account(), \ |
|||
self.mock_get_transactions(): # noqa: B950 |
|||
vals = { |
|||
"provider_ids": [(4, self.provider.id)], |
|||
"date_since": datetime(2019, 11, 4), |
|||
"date_until": datetime(2019, 11, 5), |
|||
} |
|||
wizard = self.AccountStatementPull.with_context( |
|||
active_model="account.journal", |
|||
active_id=self.journal.id, |
|||
).create(vals) |
|||
# For some reason the provider is not set in the create. |
|||
wizard.provider_ids = self.provider |
|||
wizard.action_pull() |
|||
statements = self.AccountBankStatement.search( |
|||
[("journal_id", "=", self.journal.id)] |
|||
) |
|||
new_statement = statements - initial_statement |
|||
self.assertEqual(len(new_statement.line_ids), 1) |
|||
self.assertEqual(new_statement.balance_start, 100) |
|||
self.assertEqual(new_statement.balance_end, 105.83) |
|||
# Ponto does not give balance info in transactions. |
|||
# self.assertEqual(new_statement.balance_end_real, 105.83) |
|||
|
|||
def test_ponto(self): |
|||
with self.mock_login(), \ |
|||
self.mock_synchronisation(), \ |
|||
self.mock_set_access_account(), \ |
|||
self.mock_get_transactions(): # noqa: B950 |
|||
vals = { |
|||
"provider_ids": [(4, self.provider.id)], |
|||
"date_since": datetime(2019, 11, 3), |
|||
"date_until": datetime(2019, 11, 17), |
|||
} |
|||
wizard = self.AccountStatementPull.with_context( |
|||
active_model="account.journal", |
|||
active_id=self.journal.id, |
|||
).create(vals) |
|||
# To get all the moves at once |
|||
self.provider.statement_creation_mode = "monthly" |
|||
# For some reason the provider is not set in the create. |
|||
wizard.provider_ids = self.provider |
|||
wizard.action_pull() |
|||
statement = self.AccountBankStatement.search( |
|||
[("journal_id", "=", self.journal.id)] |
|||
) |
|||
self.assertEqual(len(statement), 1) |
|||
self.assertEqual(len(statement.line_ids), 3) |
|||
sorted_amounts = sorted(statement.line_ids.mapped("amount")) |
|||
self.assertEqual(sorted_amounts, [5.48, 5.83, 6.08]) |
|||
self.assertEqual(statement.balance_end, 17.39) |
|||
# Ponto does not give balance info in transactions. |
|||
# self.assertEqual(statement.balance_end_real, 17.39) |
|||
|
|||
def test_ponto_buffer_purge(self): |
|||
"""Create some old buffer records and test purging them.""" |
|||
buffer_model = self.env["ponto.buffer"] |
|||
buffer_model.sudo()._store_transactions(self.provider, THREE_TRANSACTIONS) |
|||
# As all transactions have a different date, they will be in separate buffers. |
|||
buffers = buffer_model.search([("provider_id", "=", self.provider.id)]) |
|||
self.assertEqual(len(buffers), 3) |
|||
# Non ponto providers should not affect buffers. |
|||
self._expect_purge_result(service="dummy", retain_days=15, expected_length=3) |
|||
# If retain date not filled, buffers should not be purged. |
|||
self._expect_purge_result(expected_length=3) |
|||
# If retain date filled, buffers should be purged. |
|||
self._expect_purge_result(retain_days=15) |
|||
|
|||
def _expect_purge_result(self, service="ponto", retain_days=0, expected_length=0): |
|||
"""Check result for purge in different scenario's.""" |
|||
buffer_model = self.env["ponto.buffer"] |
|||
self.provider.write( |
|||
{ |
|||
"active": True, |
|||
"ponto_buffer_retain_days": retain_days, |
|||
"service": service, |
|||
} |
|||
) |
|||
self.provider._ponto_buffer_purge() |
|||
buffers = buffer_model.search([("provider_id", "=", self.provider.id)]) |
|||
self.assertEqual(len(buffers), expected_length) |
|||
@ -0,0 +1,150 @@ |
|||
# Copyright 2022 Therp BV <https://therp.nl>. |
|||
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl). |
|||
from dateutil.relativedelta import relativedelta |
|||
import json |
|||
|
|||
from unittest.mock import MagicMock, patch |
|||
|
|||
from odoo import fields |
|||
from odoo.tests import common |
|||
|
|||
from .test_account_statement_import_online_ponto import THREE_TRANSACTIONS |
|||
|
|||
|
|||
class TestPontoInterface(common.TransactionCase): |
|||
post_install = True |
|||
|
|||
@patch("requests.post") |
|||
def test_login(self, requests_post): |
|||
"""Check Ponto API login.""" |
|||
mock_response = MagicMock() |
|||
mock_response.status_code = 200 |
|||
mock_response.text = json.dumps( |
|||
{ |
|||
"access_token": "live_the_token", |
|||
"expires_in": 1799, |
|||
"scope": "ai", |
|||
"token_type": "bearer", |
|||
} |
|||
) |
|||
requests_post.return_value = mock_response |
|||
interface_model = self.env["ponto.interface"] |
|||
access_data = interface_model._login("uncle_john", "secret") |
|||
self.assertEqual(access_data["access_token"], "live_the_token") |
|||
self.assertIn("token_expiration", access_data) |
|||
|
|||
@patch("requests.get") |
|||
def test_set_access_account(self, requests_get): |
|||
"""Test getting account data for Ponto access.""" |
|||
mock_response = MagicMock() |
|||
mock_response.status_code = 200 |
|||
mock_response.text = json.dumps( |
|||
{ |
|||
"data": [ |
|||
{ |
|||
"id": "wrong_id", |
|||
"attributes": { |
|||
"reference": "NL66ABNA123456789", |
|||
}, |
|||
}, |
|||
{ |
|||
"id": "2ad3df83-be01-47cf-a6be-cf0de5cb4c99", |
|||
"attributes": { |
|||
"reference": "NL66RABO123456789", |
|||
}, |
|||
}, |
|||
], |
|||
} |
|||
) |
|||
requests_get.return_value = mock_response |
|||
# Start of actual test. |
|||
access_data = self._get_access_dict(include_account=False) |
|||
interface_model = self.env["ponto.interface"] |
|||
interface_model._set_access_account(access_data, "NL66RABO123456789") |
|||
self.assertIn("ponto_account", access_data) |
|||
self.assertEqual( |
|||
access_data["ponto_account"], |
|||
"2ad3df83-be01-47cf-a6be-cf0de5cb4c99" |
|||
) |
|||
|
|||
@patch("requests.post") |
|||
def test_ponto_synchronisation(self, requests_post): |
|||
"""Test requesting Ponto synchronization.""" |
|||
mock_response = MagicMock() |
|||
mock_response.status_code = 400 |
|||
mock_response.text = json.dumps( |
|||
{ |
|||
"errors": [ |
|||
{ |
|||
"code": "accountRecentlySynchronized", |
|||
"detail": |
|||
"This type of synchronization was already created recently" |
|||
" for the account. Try again later or on the Dashboard.", |
|||
"meta": {} |
|||
} |
|||
] |
|||
} |
|||
) |
|||
requests_post.return_value = mock_response |
|||
# Start of actual test (succeeds if no Exceptions occur). |
|||
access_data = self._get_access_dict() |
|||
interface_model = self.env["ponto.interface"] |
|||
interface_model._ponto_synchronisation(access_data) |
|||
|
|||
@patch("requests.get") |
|||
def test_synchronization_done(self, requests_get): |
|||
"""Test getting account data for Ponto access.""" |
|||
mock_response = MagicMock() |
|||
mock_response.status_code = 200 |
|||
mock_response.text = json.dumps({"status": "success"}) |
|||
requests_get.return_value = mock_response |
|||
# Succesfull sync. |
|||
self._check_synchronization_done(True) |
|||
# Error in sync. |
|||
mock_response.text = json.dumps({"status": "error"}) |
|||
self._check_synchronization_done(True) |
|||
# Unexpected error in sync. |
|||
mock_response.status_code = 404 |
|||
self._check_synchronization_done(False) |
|||
|
|||
def _check_synchronization_done(self, expected_result): |
|||
"""Check result for synchronization with current mock.""" |
|||
interface_model = self.env["ponto.interface"] |
|||
access_data = self._get_access_dict() |
|||
synchronization_done = interface_model._synchronization_done( |
|||
access_data, |
|||
"https//does.not.matter.com/synchronization" |
|||
) |
|||
self.assertEqual(synchronization_done, expected_result) |
|||
|
|||
@patch("requests.get") |
|||
def test_get_transactions(self, requests_get): |
|||
"""Test getting transactions from Ponto.""" |
|||
mock_response = MagicMock() |
|||
mock_response.status_code = 200 |
|||
# Key "data" will contain a list of transactions. |
|||
mock_response.text = json.dumps({"data": THREE_TRANSACTIONS}) |
|||
requests_get.return_value = mock_response |
|||
# Start of actual test. |
|||
access_data = self._get_access_dict() |
|||
interface_model = self.env["ponto.interface"] |
|||
transactions = interface_model._get_transactions(access_data, False) |
|||
self.assertEqual(len(transactions), 3) |
|||
self.assertEqual(transactions[2]["id"], "b21a6c65-1c52-4ba6-8cbc-127d2b2d85ff") |
|||
self.assertEqual( |
|||
transactions[2]["attributes"]["counterpartReference"], |
|||
"BE10325927501996" |
|||
) |
|||
|
|||
def _get_access_dict(self, include_account=True): |
|||
"""Get access dict that caches login/account information.""" |
|||
token_expiration = fields.Datetime.now() + relativedelta(seconds=1800) |
|||
access_data = { |
|||
"username": "uncle_john", |
|||
"password": "secret", |
|||
"access_token": "live_the_token", |
|||
"token_expiration": token_expiration, |
|||
} |
|||
if include_account: |
|||
access_data["ponto_account"] = "2ad3df83-be01-47cf-a6be-cf0de5cb4c99" |
|||
return access_data |
|||
@ -0,0 +1,11 @@ |
|||
<?xml version="1.0" encoding="utf-8"?> |
|||
<odoo> |
|||
<record id="action_ponto_buffer" model="ir.actions.act_window"> |
|||
<field name="name">Ponto Buffer</field> |
|||
<field name="res_model">ponto.buffer</field> |
|||
<field name="view_type">form</field> |
|||
<field name="view_mode">tree,form</field> |
|||
<field eval="False" name="view_id"/> |
|||
</record> |
|||
</odoo> |
|||
|
|||
@ -0,0 +1,13 @@ |
|||
<?xml version="1.0" encoding="utf-8"?> |
|||
<odoo> |
|||
|
|||
<menuitem |
|||
id="menu_ponto_buffer" |
|||
name="Show Ponto Buffer" |
|||
parent="account.account_reports_management_menu" |
|||
sequence="90" |
|||
action="action_ponto_buffer" |
|||
groups="account.group_account_user" |
|||
/> |
|||
|
|||
</odoo> |
|||
@ -0,0 +1,35 @@ |
|||
<?xml version="1.0" encoding="utf-8" ?> |
|||
<odoo> |
|||
<record model="ir.ui.view" id="ponto_buffer_tree"> |
|||
<field name="name">ponto.buffer.tree</field> |
|||
<field name="model">ponto.buffer</field> |
|||
<field name="arch" type="xml"> |
|||
<tree> |
|||
<field name="provider_id" /> |
|||
<field name="effective_date" /> |
|||
</tree> |
|||
</field> |
|||
</record> |
|||
|
|||
<record model="ir.ui.view" id="ponto_buffer_form"> |
|||
<field name="name">ponto.buffer.form</field> |
|||
<field name="model">ponto.buffer</field> |
|||
<field name="arch" type="xml"> |
|||
<form> |
|||
<group name="ponto_buffer_main"> |
|||
<field name="provider_id" /> |
|||
<field name="effective_date" /> |
|||
</group> |
|||
<group name="ponto_buffer_lines" string="Buffer Lines"> |
|||
<field name="buffer_line_ids" mode="tree"> |
|||
<tree> |
|||
<field name="effective_date_time" /> |
|||
<field name="ponto_id" /> |
|||
<field name="transaction_data" /> |
|||
</tree> |
|||
</field> |
|||
</group> |
|||
</form> |
|||
</field> |
|||
</record> |
|||
</odoo> |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue