Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[fix] refactor standingorders into separate class
- Loading branch information
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| """ Module for handling dkb standing orders """ | ||
| from typing import Dict, List | ||
| import logging | ||
| import requests | ||
| from dkb_robo.api import DKBRoboError | ||
|
|
||
|
|
||
| class StandingOrder: | ||
| """ StandingOrder class """ | ||
| def __init__(self, client: requests.Session, logger: logging.Logger, base_url: str = 'https://banking.dkb.de/api'): | ||
This comment has been minimized.
Sorry, something went wrong.
This comment has been minimized.
Sorry, something went wrong. 
grindsa
Author
Owner
|
||
| self.client = client | ||
| self.logger = logger | ||
| self.base_url = base_url | ||
| self.uid = None | ||
|
|
||
| def _filter(self, full_list: Dict[str, str]) -> List[Dict[str, str]]: | ||
| """ filter standing orders """ | ||
| self.logger.debug('api.StandingOrder._filter()\n') | ||
|
|
||
| so_list = [] | ||
| if 'data' in full_list: | ||
| for ele in full_list['data']: | ||
|
|
||
| try: | ||
| amount = float(ele.get('attributes', {}).get('amount', {}).get('value', None)) | ||
| except Exception as err: | ||
| self.logger.error('api.StandingOrder._filter() error: %s', err) | ||
| amount = None | ||
|
|
||
| _tmp_dic = { | ||
| 'amount': amount, | ||
| 'currencycode': ele.get('attributes', {}).get('amount', {}).get('currencyCode', None), | ||
| 'purpose': ele.get('attributes', {}).get('description', None), | ||
| 'recpipient': ele.get('attributes', {}).get('creditor', {}).get('name', None), | ||
| 'creditoraccount': ele.get('attributes', {}).get('creditor', {}).get('creditorAccount', None), | ||
| 'interval': ele.get('attributes', {}).get('recurrence', None)} | ||
| so_list.append(_tmp_dic) | ||
|
|
||
| self.logger.debug('api.StandingOrder._filter() ended with: %s entries.', len(so_list)) | ||
| return so_list | ||
|
|
||
| def fetch(self, uid) -> Dict: | ||
| """ fetch standing orders """ | ||
| self.logger.debug('api.Standorder.get()\n') | ||
|
|
||
| so_list = [] | ||
| if uid: | ||
| response = self.client.get(self.base_url + '/accounts/payments/recurring-credit-transfers' + '?accountId=' + uid) | ||
| if response.status_code == 200: | ||
| _so_list = response.json() | ||
| so_list = self._filter(_so_list) | ||
| else: | ||
| raise DKBRoboError('get_standing_orders(): account-id is required') | ||
|
|
||
| self.logger.debug('api.Wrapper.get_standing_orders() ended\n') | ||
| return so_list | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,136 @@ | ||
| # -*- coding: utf-8 -*- | ||
| # pylint: disable=r0904, c0415, c0413, r0913, w0212 | ||
| """ unittests for dkb_robo """ | ||
| import sys | ||
| import os | ||
| from datetime import date | ||
| import unittest | ||
| import logging | ||
| import json | ||
| from unittest.mock import patch, Mock, MagicMock, mock_open | ||
| from bs4 import BeautifulSoup | ||
| from mechanicalsoup import LinkNotFoundError | ||
| import io | ||
| sys.path.insert(0, '.') | ||
| sys.path.insert(0, '..') | ||
| from dkb_robo.standingorder import StandingOrder | ||
|
|
||
| def json_load(fname): | ||
| """ simple json load """ | ||
|
|
||
| with open(fname, 'r', encoding='utf8') as myfile: | ||
| data_dic = json.load(myfile) | ||
|
|
||
| return data_dic | ||
|
|
||
|
|
||
| class TestDKBRobo(unittest.TestCase): | ||
| """ test class """ | ||
|
|
||
| @patch("requests.Session") | ||
| def setUp(self, mock_session): | ||
| self.dir_path = os.path.dirname(os.path.realpath(__file__)) | ||
| self.logger = logging.getLogger('dkb_robo') | ||
| self.dkb = StandingOrder(logger=self.logger, client=mock_session) | ||
|
|
||
| @patch('dkb_robo.standingorder.StandingOrder._filter') | ||
| def test_001_fetch(self, mock_filter): | ||
| """ test StandingOrder.fetch() without uid """ | ||
| self.dkb.client = Mock() | ||
| self.dkb.client.get.return_value.status_code = 200 | ||
| self.dkb.client.get.return_value.json.return_value = {'foo': 'bar'} | ||
|
|
||
| with self.assertRaises(Exception) as err: | ||
| self.assertFalse(self.dkb.fetch(None)) | ||
| self.assertEqual('get_standing_orders(): account-id is required', str(err.exception)) | ||
| self.assertFalse(mock_filter.called) | ||
|
|
||
| @patch('dkb_robo.standingorder.StandingOrder._filter') | ||
| def test_002_fetch(self, mock_filter): | ||
| """ test StandingOrder.fetch() with uid but http error """ | ||
| self.dkb.client = Mock() | ||
| self.dkb.client.get.return_value.status_code = 400 | ||
| self.dkb.client.get.return_value.json.return_value = {'foo': 'bar'} | ||
| self.assertFalse(self.dkb.fetch(uid='uid')) | ||
| self.assertFalse(mock_filter.called) | ||
|
|
||
| @patch('dkb_robo.standingorder.StandingOrder._filter') | ||
| def test_003_fetch(self, mock_filter): | ||
| """ test StandingOrder.fetch() with uid no error """ | ||
| self.dkb.client = Mock() | ||
| self.dkb.client.get.return_value.status_code = 200 | ||
| self.dkb.client.get.return_value.json.return_value = {'foo': 'bar'} | ||
| mock_filter.return_value = 'mock_filter' | ||
| self.assertEqual('mock_filter', self.dkb.fetch(uid='uid')) | ||
| self.assertTrue(mock_filter.called) | ||
|
|
||
| def test_004__filter(self): | ||
| """ test StandingOrder._filter() with empty list """ | ||
| full_list = {} | ||
| self.assertFalse(self.dkb._filter(full_list)) | ||
|
|
||
| def test_005__filter(self): | ||
| """ test StandingOrder._filter() with list """ | ||
| full_list = { | ||
| "data": [ | ||
| { | ||
| "attributes": { | ||
| "description": "description", | ||
| "amount": { | ||
| "currencyCode": "EUR", | ||
| "value": "100.00" | ||
| }, | ||
| "creditor": { | ||
| "name": "cardname", | ||
| "creditorAccount": { | ||
| "iban": "crediban", | ||
| "bic": "credbic" | ||
| } | ||
| }, | ||
| "recurrence": { | ||
| "from": "2020-01-01", | ||
| "until": "2025-12-01", | ||
| "frequency": "monthly", | ||
| "nextExecutionAt": "2020-02-01" | ||
| } | ||
| } | ||
| }]} | ||
| result = [{'amount': 100.0, 'currencycode': 'EUR', 'purpose': 'description', 'recpipient': 'cardname', 'creditoraccount': {'iban': 'crediban', 'bic': 'credbic'}, 'interval': {'from': '2020-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'nextExecutionAt': '2020-02-01'}}] | ||
| self.assertEqual(result, self.dkb._filter(full_list)) | ||
|
|
||
| def test_006__filter(self): | ||
| """ test StandingOrder._filter() with list from file """ | ||
| so_list = json_load(self.dir_path + '/mocks/so.json') | ||
| result = [{'amount': 100.0, 'currencycode': 'EUR', 'purpose': 'description1', 'recpipient': 'name1', 'creditoraccount': {'iban': 'iban1', 'bic': 'bic1'}, 'interval': {'from': '2022-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-11-01'}}, {'amount': 200.0, 'currencycode': 'EUR', 'purpose': 'description2', 'recpipient': 'name2', 'creditoraccount': {'iban': 'iban2', 'bic': 'bic2'}, 'interval': {'from': '2022-02-01', 'until': '2025-12-02', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-11-02'}}, {'amount': 300.0, 'currencycode': 'EUR', 'purpose': 'description3', 'recpipient': 'name3', 'creditoraccount': {'iban': 'iban3', 'bic': 'bic3'}, 'interval': {'from': '2022-03-01', 'until': '2025-03-01', 'frequency': 'monthly', 'holidayExecutionStrategy': 'following', 'nextExecutionAt': '2022-03-01'}}] | ||
| self.assertEqual(result, self.dkb._filter(so_list)) | ||
|
|
||
| def test_007__filter(self): | ||
| """ test StandingOrder._filter() with incomplete list """ | ||
| full_list = { | ||
| "data": [ | ||
| { | ||
| "attributes": { | ||
| "description": "description", | ||
| "creditor": { | ||
| "name": "cardname", | ||
| "creditorAccount": { | ||
| "iban": "crediban", | ||
| "bic": "credbic" | ||
| } | ||
| }, | ||
| "recurrence": { | ||
| "from": "2020-01-01", | ||
| "until": "2025-12-01", | ||
| "frequency": "monthly", | ||
| "nextExecutionAt": "2020-02-01" | ||
| } | ||
| } | ||
| }]} | ||
| result = [{'amount': None, 'currencycode': None, 'purpose': 'description', 'recpipient': 'cardname', 'creditoraccount': {'iban': 'crediban', 'bic': 'credbic'}, 'interval': {'from': '2020-01-01', 'until': '2025-12-01', 'frequency': 'monthly', 'nextExecutionAt': '2020-02-01'}}] | ||
| with self.assertLogs('dkb_robo', level='INFO') as lcm: | ||
| self.assertEqual(result, self.dkb._filter(full_list)) | ||
| self.assertIn("ERROR:dkb_robo:api.StandingOrder._filter() error: float() argument must be a string or a real number, not 'NoneType'", lcm.output) | ||
|
|
||
| if __name__ == '__main__': | ||
|
|
||
| unittest.main() |
I think it is not needed to pass the logger instance around. Instead, you can just create an instance per module(
logger = logging.getLogger(__name__)) and even could configure the logging formatter to output the module/logger name -- then you don't need to prefix the log messages anymore. https://docs.python.org/3/howto/logging.html#configuring-logging