From 7c74c154637c23a5d4e9705f03ad9e84cba9f19a Mon Sep 17 00:00:00 2001 From: "Ronald Portier (Therp BV)" Date: Wed, 31 Aug 2022 12:34:31 +0200 Subject: [PATCH] [IMP] *_online_ponto: black Blacked code for better comparison / easier backport from later versions. --- .pre-commit-config.yaml | 1 - .../__manifest__.py | 4 +- .../online_bank_statement_provider_ponto.py | 191 ++++++++-------- ...ount_bank_statement_import_online_ponto.py | 203 ++++++++++-------- 4 files changed, 224 insertions(+), 175 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index ccc8891..49ed876 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -38,7 +38,6 @@ repos: rev: v3.4.1 hooks: - id: flake8 - language_version: python3.6 name: flake8 excluding __init__.py exclude: __init__\.py - repo: https://github.com/pre-commit/mirrors-pylint diff --git a/account_bank_statement_import_online_ponto/__manifest__.py b/account_bank_statement_import_online_ponto/__manifest__.py index dcfc8a2..1b0b849 100644 --- a/account_bank_statement_import_online_ponto/__manifest__.py +++ b/account_bank_statement_import_online_ponto/__manifest__.py @@ -9,7 +9,5 @@ "license": "AGPL-3", "installable": True, "depends": ["account_bank_statement_import_online"], - "data": [ - "view/online_bank_statement_provider.xml" - ], + "data": ["view/online_bank_statement_provider.xml"], } diff --git a/account_bank_statement_import_online_ponto/models/online_bank_statement_provider_ponto.py b/account_bank_statement_import_online_ponto/models/online_bank_statement_provider_ponto.py index 6802313..ed5416e 100644 --- a/account_bank_statement_import_online_ponto/models/online_bank_statement_provider_ponto.py +++ b/account_bank_statement_import_online_ponto/models/online_bank_statement_provider_ponto.py @@ -14,28 +14,28 @@ 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' +PONTO_ENDPOINT = "https://api.myponto.com" class OnlineBankStatementProviderPonto(models.Model): - _inherit = 'online.bank.statement.provider' + _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) def ponto_reset_last_identifier(self): - self.write({'ponto_last_identifier': False}) + self.write({"ponto_last_identifier": False}) @api.model def _get_available_services(self): return super()._get_available_services() + [ - ('ponto', 'MyPonto.com'), + ("ponto", "MyPonto.com"), ] def _obtain_statement_data(self, date_since, date_until): self.ensure_one() - if self.service != 'ponto': + if self.service != "ponto": return super()._obtain_statement_data( date_since, date_until, @@ -49,126 +49,145 @@ class OnlineBankStatementProviderPonto(models.Model): 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.')) + 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 ( + 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) + access_token = data.get("access_token", False) if not access_token: - raise UserError(_('Ponto : no 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)) + 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, } + 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()) + 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', []): + for account in data.get("data", []): iban = sanitize_account_number( - account.get('attributes', {}).get('reference', '')) - res[iban] = account.get('id') + account.get("attributes", {}).get("reference", "") + ) + res[iban] = account.get("id") return res - raise UserError(_('%s \n\n %s') % (response.status_code, response.text)) + 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' + 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) + } + 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) + sync_id = data.get("attributes", {}).get("resourceId", False) else: - raise UserError(_('Error during Create Synchronisation %s \n\n %s') % ( - response.status_code, response.text)) + 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 + 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'): + 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_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 + 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()) + 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') + if params.get("before"): + params.pop("before") data = json.loads(response.text) - links = data.get('links', {}) + links = data.get("links", {}) if page_next: - page_url = links.get('next', False) + page_url = links.get("next", False) else: - page_url = links.get('prev', False) - transactions = data.get('data', []) + 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')) + 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') + 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)) + _("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 @@ -177,10 +196,8 @@ class OnlineBankStatementProviderPonto(models.Model): """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') - ) + 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): @@ -191,30 +208,36 @@ class OnlineBankStatementProviderPonto(models.Model): account_id = account_ids.get(iban) if not account_id: raise UserError( - _('Ponto : wrong configuration, unknow account %s') - % journal.bank_account_id.acc_number) + _("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) + account_id, date_since, date_until + ) new_transactions = [] sequence = 0 for transaction in transaction_lines: sequence += 1 - attributes = transaction.get('attributes', {}) - ref_list = [attributes.get(x) for x in { - "description", - "counterpartName", - "counterpartReference", - } if attributes.get(x)] + 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')) + 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'], + "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"] diff --git a/account_bank_statement_import_online_ponto/tests/test_account_bank_statement_import_online_ponto.py b/account_bank_statement_import_online_ponto/tests/test_account_bank_statement_import_online_ponto.py index 8b4353b..f88df80 100644 --- a/account_bank_statement_import_online_ponto/tests/test_account_bank_statement_import_online_ponto.py +++ b/account_bank_statement_import_online_ponto/tests/test_account_bank_statement_import_online_ponto.py @@ -7,119 +7,148 @@ from unittest import mock from odoo import fields from odoo.tests import common -_module_ns = 'odoo.addons.account_bank_statement_import_online_ponto' +_module_ns = "odoo.addons.account_bank_statement_import_online_ponto" _provider_class = ( _module_ns - + '.models.online_bank_statement_provider_ponto' - + '.OnlineBankStatementProviderPonto' + + ".models.online_bank_statement_provider_ponto" + + ".OnlineBankStatementProviderPonto" ) -class TestAccountBankAccountStatementImportOnlineQonto( - common.TransactionCase -): - +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.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, - }) + { + "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--'}, + _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'}, + _provider_class + "._ponto_get_account_ids", + return_value={"FR0214508000302245362775K46": "id"}, ) self.mock_synchronisation = lambda: mock.patch( - _provider_class + '._ponto_synchronisation', + _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}}], + _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(): + 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),