From 923085c3a0c159903a7f4c9a8b5c71a4e4c068e3 Mon Sep 17 00:00:00 2001 From: grindsa Date: Thu, 26 Dec 2024 19:07:39 +0100 Subject: [PATCH] [fix] refactor transactions into a separate class --- dkb_robo/api.py | 359 ----------------------------- dkb_robo/dkb_robo.py | 11 +- dkb_robo/standingorder.py | 10 +- dkb_robo/transaction.py | 340 +++++++++++++++++++++++++++ test/test_api.py | 296 ------------------------ test/test_dkb_robo.py | 17 +- test/test_postbox.py | 6 +- test/test_standingorder.py | 2 +- test/test_transaction.py | 460 +++++++++++++++++++++++++++++++++++++ 9 files changed, 825 insertions(+), 676 deletions(-) create mode 100644 dkb_robo/transaction.py create mode 100644 test/test_transaction.py diff --git a/dkb_robo/api.py b/dkb_robo/api.py index 2a41506..0a73d5a 100644 --- a/dkb_robo/api.py +++ b/dkb_robo/api.py @@ -54,94 +54,6 @@ def __init__(self, dkb_user: str = None, dkb_password: str = None, chip_tan: boo except (ValueError, TypeError): self.mfa_device = 0 - def _add_account_transactionamount(self, transaction: Dict[str, str]) -> Dict[str, str]: - """ add amount from accont transaction """ - self.logger.debug('api.Wrapper._add_account_transactionamount()\n') - - output_dic = {} - if 'amount' in transaction['attributes']: - if 'value' in transaction['attributes']['amount']: - output_dic['amount'] = float(transaction['attributes']['amount']['value']) - if 'currencyCode' in transaction['attributes']['amount']: - output_dic['currencycode'] = transaction['attributes']['amount']['currencyCode'] - - self.logger.debug('api.Wrapper._add_account_transactionamount() ended\n') - return output_dic - - def _add_account_transaction_creditorinfo(self, transaction: Dict[str, str]) -> Dict[str, str]: - """ we need creditor information for outgoing payments""" - self.logger.debug('api.Wrapper._add_account_transaction_creditorinfo()\n') - - output_dic = {} - if 'creditor' in transaction['attributes']: - if 'creditorAccount' in transaction['attributes']['creditor'] and 'iban' in transaction['attributes']['creditor']['creditorAccount']: - output_dic['peeraccount'] = transaction['attributes']['creditor']['creditorAccount']['iban'] - if 'agent' in transaction['attributes']['creditor'] and 'bic' in transaction['attributes']['creditor']['agent']: - output_dic['peerbic'] = transaction['attributes']['creditor']['agent']['bic'] - if 'name' in transaction['attributes']['creditor']: - output_dic['peer'] = transaction['attributes']['creditor']['name'] - if 'id' in transaction['attributes']['creditor']: - output_dic['peerid'] = transaction['attributes']['creditor']['id'] - else: - output_dic['peerid'] = '' - - self.logger.debug('api.Wrapper._add_account_transaction_creditorinfo() ended\n') - return output_dic - - def _add_account_transaction_debtorinfo(self, transaction: Dict[str, str]) -> Dict[str, str]: - """we need debitor information for incoming payments """ - self.logger.debug('api.Wrapper._add_account_transaction_debtorinfo()\n') - - output_dic = {} - if 'debtor' in transaction['attributes']: - if 'debtorAccount' in transaction['attributes']['debtor'] and 'iban' in transaction['attributes']['debtor']['debtorAccount']: - output_dic['peeraccount'] = transaction['attributes']['debtor']['debtorAccount']['iban'] - if 'agent' in transaction['attributes']['debtor'] and 'bic' in transaction['attributes']['debtor']['agent']: - output_dic['peerbic'] = transaction['attributes']['debtor']['agent']['bic'] - if 'intermediaryName' in transaction['attributes']['debtor'] and transaction['attributes']['debtor']['intermediaryName']: - output_dic['peer'] = transaction['attributes']['debtor']['intermediaryName'] - else: - if 'name' in transaction['attributes']['debtor']: - output_dic['peer'] = transaction['attributes']['debtor']['name'] - - # add perrid - output_dic['peerid'] = self._add_account_transaction_debtorpeerid(transaction) - - self.logger.debug('api.Wrapper._add_account_transaction_debtorinfo() ended\n') - return output_dic - - def _add_account_transaction_debtorpeerid(self, transaction: Dict[str, str]) -> str: - """ lookup peerid """ - self.logger.debug('api.Wrapper._add_account_transaction_debtorpeerid()\n') - peer_id = '' - if 'id' in transaction['attributes']['debtor']: - peer_id = transaction['attributes']['debtor']['id'] - - self.logger.debug('api.Wrapper._add_account_transaction_debtorpeerid()\n') - return peer_id - - def _add_account_transactioninformation(self, transaction: Dict[str, str]) -> Dict[str, str]: - """ add infromation from accont transaction """ - self.logger.debug('api.Wrapper._add_account_transactioninformation()\n') - - output_dic = {} - if 'bookingDate' in transaction['attributes']: - output_dic['date'] = transaction['attributes']['bookingDate'] - output_dic['bdate'] = transaction['attributes']['bookingDate'] - if 'valueDate' in transaction['attributes']: - output_dic['vdate'] = transaction['attributes']['valueDate'] - if 'endToEndId' in transaction['attributes']: - output_dic['customerreference'] = transaction['attributes']['endToEndId'] - if 'mandateId' in transaction['attributes']: - output_dic['mandatereference'] = transaction['attributes']['mandateId'] - if 'transactionType' in transaction['attributes']: - output_dic['postingtext'] = transaction['attributes']['transactionType'] - if 'description' in transaction['attributes']: - output_dic['reasonforpayment'] = transaction['attributes']['description'] - - self.logger.debug('api.Wrapper._add_account_transactioninformation() ended\n') - return output_dic - def _get_account_details(self, aid, accounts_dic: Dict[str, str]) -> Dict[str, str]: """ get credit account details from cc json """ self.logger.debug('api.Wrapper._get_account_details(%s)\n', aid) @@ -207,64 +119,6 @@ def _add_accountname(self, account: Dict[str, str]) -> Dict[str, str]: self.logger.debug('api.Wrapper._add_accountname() ended\n') return output_dic - def _add_brokerage_informationy(self, position: Dict[str, str]) -> Dict[str, str]: - """ add lastorder date and value """ - self.logger.debug('api.Wrapper._add_brokerage_informationy()\n') - - output_dic = {} - if 'lastOrderDate' in position['attributes']: - output_dic['lastorderdate'] = position['attributes']['lastOrderDate'] - - if 'performance' in position['attributes'] and 'currentValue' in position['attributes']['performance'] and 'value' in position['attributes']['performance']['currentValue']: - output_dic['price_euro'] = position['attributes']['performance']['currentValue']['value'] - - self.logger.debug('api.Wrapper._add_brokerage_informationy() ended\n') - return output_dic - - def _add_brokerage_instrumentinformation(self, ele: Dict[str, str]) -> Dict[str, str]: - """ add instrument information """ - self.logger.debug('api.Wrapper._add_brokerage_instrumentinformation()\n') - - output_dic = {} - if 'attributes' in ele: - if 'name' in ele['attributes'] and 'short' in ele['attributes']['name']: - output_dic['text'] = ele['attributes']['name']['short'] - if 'identifiers' in ele['attributes']: - for identifier in ele['attributes']['identifiers']: - if identifier['identifier'] == 'isin': - output_dic['isin_wkn'] = identifier['value'] - break - - self.logger.debug('api.Wrapper._add_brokerage_instrumentinformation() ended\n') - return output_dic - - def _add_brokerage_quoteinformation(self, ele: Dict[str, str]) -> Dict[str, str]: - """ add quote information """ - self.logger.debug('api.Wrapper._add_brokerage_quoteinformation()\n') - - output_dic = {} - if 'attributes' in ele and 'price' in ele['attributes']: - if 'value' in ele['attributes']['price']: - output_dic['price'] = float(ele['attributes']['price']['value']) - if 'currencyCode' in ele['attributes']['price']: - output_dic['currencycode'] = ele['attributes']['price']['currencyCode'] - if 'market' in ele['attributes']: - output_dic['market'] = ele['attributes']['market'] - - self.logger.debug('api.Wrapper._add_brokerage_quoteinformation() ended\n') - return output_dic - - def _add_brokerage_quantity(self, position: Dict[str, str]) -> Dict[str, str]: - """ add quantity information """ - self.logger.debug('api.Wrapper._add_brokerage_quantity()\n') - output_dic = {} - if 'quantity' in position['attributes']: - output_dic['shares'] = position['attributes']['quantity']['value'] - output_dic['quantity'] = float(position['attributes']['quantity']['value']) - output_dic['shares_unit'] = position['attributes']['quantity']['unit'] - - self.logger.debug('api.Wrapper._add_brokerage_quantity() ended\n') - return output_dic def _add_brokerageholder(self, depot: Dict[str, str]) -> Dict[str, str]: """ add card holder information """ @@ -305,18 +159,7 @@ def _add_brokerageperformance(self, depot: Dict[str, str]) -> Dict[str, str]: self.logger.debug('api.Wrapper._add_brokerageperformance() ended\n') return output_dic - def _add_card_transactionamount(self, transaction: Dict[str, str]) -> Dict[str, str]: - """ add amount from card transaction """ - self.logger.debug('api.Wrapper._add_card_transactionamount()\n') - output_dic = {} - if 'amount' in transaction['attributes']: - if 'value' in transaction['attributes']['amount']: - output_dic['amount'] = float(transaction['attributes']['amount']['value']) - if 'currencyCode' in transaction['attributes']['amount']: - output_dic['currencycode'] = transaction['attributes']['amount']['currencyCode'] - self.logger.debug('api.Wrapper._add_card_transactionamount() ended\n') - return output_dic def _add_cardbalance(self, card: Dict[str, str]) -> Dict[str, str]: """ add card balance to dictionary """ @@ -384,19 +227,6 @@ def _add_cardlimit(self, card: Dict[str, str]) -> Dict[str, str]: self.logger.debug('api.Wrapper._add_cardlimit() ended\n') return output_dic - def _add_card_transactioninformation(self, transaction: Dict[str, str]) -> Dict[str, str]: - """ add card transaction infromatoin """ - self.logger.debug('api.Wrapper._add_card_transactioninformation()\n') - output_dic = {} - if 'bookingDate' in transaction['attributes']: - output_dic['bdate'] = transaction['attributes']['bookingDate'] - output_dic['vdate'] = transaction['attributes']['bookingDate'] - if 'description' in transaction['attributes']: - output_dic['text'] = transaction['attributes']['description'] - - self.logger.debug('api.Wrapper._add_card_transactioninformation() ended\n') - return output_dic - def _add_cardname(self, card: Dict[str, str]) -> Dict[str, str]: """ add card name """ self.logger.debug('api.Wrapper._add_cardname()\n') @@ -682,88 +512,6 @@ def _do_sso_redirect(self): - def _filter_transactions(self, transaction_list: List[Dict[str, str]], date_from: str, date_to: str, transaction_type: str) -> List[Dict[str, str]]: - """ filter transaction by date """ - self.logger.debug('api.Wrapper._filter_transactions()\n') - - # support transation type 'reserved' for backwards compatibility - transaction_type = 'pending' if transaction_type == 'reserved' else transaction_type - - try: - date_from_uts = int(time.mktime(datetime.datetime.strptime(date_from, LEGACY_DATE_FORMAT).timetuple())) - except ValueError: - date_from_uts = int(time.mktime(datetime.datetime.strptime(date_from, API_DATE_FORMAT).timetuple())) - - try: - date_to_uts = int(time.mktime(datetime.datetime.strptime(date_to, LEGACY_DATE_FORMAT).timetuple())) - except ValueError: - date_to_uts = int(time.mktime(datetime.datetime.strptime(date_to, API_DATE_FORMAT).timetuple())) - - filtered_transaction_list = [] - for transaction in transaction_list: - if 'attributes' in transaction and 'status' in transaction['attributes'] and 'bookingDate' in transaction['attributes']: - if transaction['attributes']['status'] == transaction_type: - bookingdate_uts = int(time.mktime(datetime.datetime.strptime(transaction['attributes']['bookingDate'], API_DATE_FORMAT).timetuple())) - if date_from_uts <= bookingdate_uts <= date_to_uts: - filtered_transaction_list.append(transaction) - - self.logger.debug('api.Wrapper._filter_transactions() ended\n') - return filtered_transaction_list - - def _format_account_transactions(self, transaction_list: List[Dict[str, str]]) -> List[Dict[str, str]]: - """ format transactions """ - self.logger.debug('api.Wrapper._format_transactions()\n') - - format_transaction_list = [] - for transaction in transaction_list: - if 'attributes' in transaction: - transaction_dic = {**self._add_account_transactionamount(transaction), **self._add_account_transactioninformation(transaction)} - - if transaction_dic['amount'] > 0: - # incoming payment - collect debitor information - transaction_dic = {**transaction_dic, **self._add_account_transaction_debtorinfo(transaction)} - else: - # outgoing payment - collect creditor information - transaction_dic = {**transaction_dic, **self._add_account_transaction_creditorinfo(transaction)} - - # add posting test for backwards compability - if 'postingtext' in transaction_dic and 'peer' in transaction_dic and 'reasonforpayment' in transaction_dic: - transaction_dic['text'] = f'{transaction_dic["postingtext"]} {transaction_dic["peer"]} {transaction_dic["reasonforpayment"]}' - - format_transaction_list.append(transaction_dic) - - self.logger.debug('api.Wrapper._format_account_transactions() ended\n') - return format_transaction_list - - def _format_brokerage_account(self, brokerage_dic: Dict[str, str]) -> List[Dict[str, str]]: - """ format brokerage dictionary """ - self.logger.debug('api.Wrapper._format_brokerage_account(%s)\n', len(brokerage_dic)) - - position_list = [] - included_list = self._get_brokerage_includedlist(brokerage_dic) - - if 'data' in brokerage_dic: - for position in brokerage_dic['data']: - position_dic = self._get_brokerage_position(position, included_list) - if position_dic: - position_list.append(position_dic) - - self.logger.debug('api.Wrapper._format_brokerage_account() ended\n') - return position_list - - def _format_card_transactions(self, transaction_list: List[Dict[str, str]]) -> List[Dict[str, str]]: - """ format credit card transactions """ - self.logger.debug('api.Wrapper._format_card_transactions(%s)\n', len(transaction_list)) - - format_transaction_list = [] - for transaction in transaction_list: - if 'attributes' in transaction: - transaction_dic = {**self._add_card_transactionamount(transaction), **self._add_card_transactioninformation(transaction)} - format_transaction_list.append(transaction_dic) - - self.logger.debug('api.Wrapper._format_card_transactions() ended\n') - return format_transaction_list - def _get_accounts(self) -> Dict[str, str]: """ get accounts via API """ self.logger.debug('api.Wrapper._get_accounts()\n') @@ -806,33 +554,6 @@ def _get_brokerage_details(self, bid: str, brokerage_dic: Dict[str, str]) -> Dic self.logger.debug('api.Wrapper._get_brokerage_details() ended\n') return output_dic - def _get_brokerage_includedlist(self, brokerage_dic: Dict[str, str]) -> Dict[str, str]: - """ get include list from brokerage_account dictionary""" - self.logger.debug('api.Wrapper._get_brokerage_includedlist()\n') - - included_list = [] - if 'included' in brokerage_dic: - included_list = brokerage_dic['included'] - - self.logger.debug('api.Wrapper._get_brokerage_includedlist() ended\n') - return included_list - - def _get_brokerage_position(self, position: Dict[str, str], included_list: List[Dict[str, str]]): - """ get information on a single position witin a brokerage account """ - self.logger.debug('api.Wrapper._get_brokerage_position()\n') - (_instrument_id, _quote_id) = self._get_relationship_ids(position) - - position_dic = {} - if 'attributes' in position: - position_dic = {**self._add_brokerage_quantity(position), **self._add_brokerage_informationy(position)} - for ele in included_list: - if 'id' in ele and ele['id'] == _instrument_id: - position_dic = {**position_dic, **self._add_brokerage_instrumentinformation(ele)} - if 'id' in ele and ele['id'] == _quote_id: - position_dic = {**position_dic, **self._add_brokerage_quoteinformation(ele)} - - self.logger.debug('api.Wrapper._get_brokerage_position() ended\n') - return position_dic def _get_cards(self) -> Dict[str, str]: """ get cards via API """ @@ -952,21 +673,6 @@ def _get_overview(self) -> Dict[str, str]: self.logger.debug('api.Wrapper._get_overview() ended\n') return self._build_account_dic(portfolio_dic) - def _get_relationship_ids(self, position: Dict[str, str]) -> Tuple[str, str]: - """ get relationship ids from depot position """ - self.logger.debug('api.Wrapper._get_relationship_ids()\n') - - instrument_id = None - quote_id = None - if 'relationships' in position: - if 'instrument' in position['relationships'] and 'data' in position['relationships']['instrument'] and 'id' in position['relationships']['instrument']['data']: - instrument_id = position['relationships']['instrument']['data']['id'] - if 'quote' in position['relationships'] and 'data' in position['relationships']['quote'] and 'id' in position['relationships']['quote']['data']: - quote_id = position['relationships']['quote']['data']['id'] - - self.logger.debug('api.Wrapper._get_relationship_ids()\n') - return instrument_id, quote_id - def _get_token(self): """ get access token """ self.logger.debug('api.Wrapper._get_token()\n') @@ -1123,71 +829,6 @@ def get_credit_limits(self) -> Dict[str, str]: self.logger.debug('api.Wrapper.get_credit_limits() ended\n') return limit_dic - def _get_transaction_url(self, tr_dic): - """ get transaction url """ - self.logger.debug('api.Wrapper._get_transaction_url()\n') - - transaction_url = None - if 'links' in tr_dic and 'next' in tr_dic['links']: - self.logger.debug('api.Wrapper._get_transactions(): next page: %s', tr_dic['links']['next']) - transaction_url = self.base_url + self.api_prefix + '/accounts' + tr_dic['links']['next'] - else: - self.logger.debug('api.Wrapper._get_transactions(): no next page') - transaction_url = None - - self.logger.debug('api.Wrapper._get_transaction_url() ended\n') - return transaction_url - - def _get_transaction_list(self, transaction_url: str) -> Dict[str, str]: - """ get transaction list""" - self.logger.debug('api.Wrapper._get_transaction_list(%s)\n', transaction_url) - - transaction_dic = {'data': [], 'included': []} - while transaction_url: - response = self.client.get(transaction_url) - if response.status_code == 200: - _transaction_dic = response.json() - if 'data' in _transaction_dic: - transaction_dic['data'].extend(_transaction_dic['data']) - transaction_url = self._get_transaction_url(_transaction_dic) # get next page - - else: - self.logger.debug('api.Wrapper._get_transactions(): no data in response') - transaction_url = None - - if 'included' in _transaction_dic: - transaction_dic['included'].extend(_transaction_dic['included']) - else: - self.logger.error('api.Wrapper._get_transactions(): RC is not 200 but %s', response.status_code) - break - - self.logger.debug('api.Wrapper._get_transaction_list() ended with %s entries\n', len(transaction_dic['data'])) - return transaction_dic - - def get_transactions(self, transaction_url: str, atype: str, date_from: str, date_to: str, transaction_type: str) -> List[Dict[str, str]]: - """ get transactions via API """ - self.logger.debug('api.Wrapper.get_transactions(%s, %s)\n', atype, transaction_type) - - transaction_list = [] - - if transaction_url and atype != 'depot': - transaction_url = transaction_url + '?filter[bookingDate][GE]=' + date_from + '&filter[bookingDate][LE]=' + date_to + '&expand=Merchant&page[size]=400' - - transaction_dic = self._get_transaction_list(transaction_url) - - if transaction_dic and 'data' in transaction_dic and len(transaction_dic['data']) > 0: - if atype == 'account': - transaction_list = self._filter_transactions(transaction_dic['data'], date_from, date_to, transaction_type) - transaction_list = self._format_account_transactions(transaction_list) - elif atype == 'creditcard': - transaction_list = self._filter_transactions(transaction_dic['data'], date_from, date_to, transaction_type) - transaction_list = self._format_card_transactions(transaction_list) - elif atype == 'depot': - transaction_list = self._format_brokerage_account(transaction_dic) - - self.logger.debug('api.Wrapper.get_transactions() ended\n') - return transaction_list - def login(self) -> Tuple[Dict, None]: """ login into DKB banking area and perform an sso redirect """ self.logger.debug('api.Wrapper.login()\n') diff --git a/dkb_robo/dkb_robo.py b/dkb_robo/dkb_robo.py index 5129d76..3a10737 100644 --- a/dkb_robo/dkb_robo.py +++ b/dkb_robo/dkb_robo.py @@ -3,8 +3,9 @@ # -*- coding: utf-8 -*- from pathlib import Path import time -from dkb_robo.standingorder import StandingOrder from dkb_robo.postbox import PostBox +from dkb_robo.standingorder import StandingOrder +from dkb_robo.transaction import Transaction from dkb_robo.utilities import logger_setup, validate_dates, get_dateformat from dkb_robo.api import Wrapper @@ -89,8 +90,8 @@ def get_transactions(self, transaction_url, atype, date_from, date_to, transacti self.logger.debug('DKBRobo.get_transactions(%s/%s: %s/%s)\n', transaction_url, atype, date_from, date_to) (date_from, date_to) = validate_dates(self.logger, date_from, date_to) - - transaction_list = self.wrapper.get_transactions(transaction_url, atype, date_from, date_to, transaction_type) + transaction = Transaction(client=self.wrapper.client, logger=self.logger) + transaction_list = transaction.get(transaction_url, atype, date_from, date_to, transaction_type) self.logger.debug('DKBRobo.get_transactions(): %s transactions returned\n', len(transaction_list)) return transaction_list @@ -104,7 +105,7 @@ def download(self, path: Path, download_all: bool, prepend_date: bool = False, m """ download postbox documents """ if path is None: list_only = True - postbox = PostBox(client = self.wrapper.client, logger = self.logger) + postbox = PostBox(client=self.wrapper.client, logger=self.logger) documents = postbox.fetch_items() if not download_all: @@ -123,7 +124,7 @@ def download(self, path: Path, download_all: bool, prepend_date: bool = False, m if not list_only: self.logger.info("Downloading %s to %s...", doc.subject(), target) - if doc.download(self.wrapper.client, target/filename): + if doc.download(self.wrapper.client, target / filename): if mark_read: doc.mark_read(self.wrapper.client, True) time.sleep(.5) diff --git a/dkb_robo/standingorder.py b/dkb_robo/standingorder.py index 118f924..3f9113a 100644 --- a/dkb_robo/standingorder.py +++ b/dkb_robo/standingorder.py @@ -15,7 +15,7 @@ def __init__(self, client: requests.Session, logger: logging.Logger, base_url: s def _filter(self, full_list: Dict[str, str]) -> List[Dict[str, str]]: """ filter standing orders """ - self.logger.debug('api.StandingOrder._filter()\n') + self.logger.debug('standing.StandingOrder._filter()\n') so_list = [] if 'data' in full_list: @@ -36,12 +36,12 @@ def _filter(self, full_list: Dict[str, str]) -> List[Dict[str, str]]: '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)) + self.logger.debug('standing.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') + self.logger.debug('standing.StandingOrder.fetch()\n') so_list = [] if uid: @@ -50,7 +50,7 @@ def fetch(self, uid) -> Dict: _so_list = response.json() so_list = self._filter(_so_list) else: - raise DKBRoboError('get_standing_orders(): account-id is required') + raise DKBRoboError('account-id is required to fetch standing orders') - self.logger.debug('api.Wrapper.get_standing_orders() ended\n') + self.logger.debug('standing.StandingOrder.fetch() ended\n') return so_list diff --git a/dkb_robo/transaction.py b/dkb_robo/transaction.py new file mode 100644 index 0000000..e62a3a8 --- /dev/null +++ b/dkb_robo/transaction.py @@ -0,0 +1,340 @@ +""" Module for handling dkb transactions """ +# pylint: disable=c0415, r0913 +import datetime +import time +from typing import Dict, List +import logging +import requests +from dkb_robo.api import DKBRoboError +from dkb_robo.utilities import get_dateformat + +LEGACY_DATE_FORMAT, API_DATE_FORMAT = get_dateformat() + + +class Transaction: + """ Transaction class """ + + def __init__(self, client: requests.Session, logger: logging.Logger, base_url: str = 'https://banking.dkb.de/api'): + self.client = client + self.logger = logger + self.base_url = base_url + self.uid = None + + def _nextpage_url(self, tr_dic): + """ get transaction url """ + self.logger.debug('transaction.Transaction._nextpage_url()\n') + + transaction_url = None + if 'links' in tr_dic and 'next' in tr_dic['links']: + self.logger.debug('transaction.Transaction._nextpage_url(): next page: %s', tr_dic['links']['next']) + transaction_url = self.base_url + '/accounts' + tr_dic['links']['next'] + else: + self.logger.debug('transaction.Transaction._nextpage_url(): no next page') + transaction_url = None + + self.logger.debug('transaction.Transaction._nextpage_url() ended\n') + return transaction_url + + def _fetch(self, transaction_url: str) -> Dict[str, str]: + """ get transaction list""" + self.logger.debug('transaction.Transaction.fetch(%s)\n', transaction_url) + + transaction_dic = {'data': [], 'included': []} + while transaction_url: + response = self.client.get(transaction_url) + if response.status_code == 200: + _transaction_dic = response.json() + if 'data' in _transaction_dic: + transaction_dic['data'].extend(_transaction_dic['data']) + transaction_url = self._nextpage_url(_transaction_dic) # get next page + else: + self.logger.debug('fetch transactions: no data in response') + transaction_url = None + + if 'included' in _transaction_dic: + transaction_dic['included'].extend(_transaction_dic['included']) + else: + self.logger.error('fetch transactions: http status code is not 200 but %s', response.status_code) + break + + self.logger.debug('transaction.Transaction.fetch() ended with %s entries\n', len(transaction_dic['data'])) + return transaction_dic + + def _filter(self, transaction_list: List[Dict[str, str]], date_from: str, date_to: str, transaction_type: str) -> List[Dict[str, str]]: + """ filter transactions """ + self.logger.debug('transaction.Transaction._filter()\n') + + # support transation type 'reserved' for backwards compatibility + transaction_type = 'pending' if transaction_type == 'reserved' else transaction_type + + try: + date_from_uts = int(time.mktime(datetime.datetime.strptime(date_from, LEGACY_DATE_FORMAT).timetuple())) + except ValueError: + date_from_uts = int(time.mktime(datetime.datetime.strptime(date_from, API_DATE_FORMAT).timetuple())) + + try: + date_to_uts = int(time.mktime(datetime.datetime.strptime(date_to, LEGACY_DATE_FORMAT).timetuple())) + except ValueError: + date_to_uts = int(time.mktime(datetime.datetime.strptime(date_to, API_DATE_FORMAT).timetuple())) + + filtered_transaction_list = [] + for transaction in transaction_list: + # print(transaction) + if 'attributes' in transaction and 'status' in transaction['attributes'] and 'bookingDate' in transaction['attributes']: + if transaction['attributes']['status'] == transaction_type: + bookingdate_uts = int(time.mktime(datetime.datetime.strptime(transaction['attributes']['bookingDate'], API_DATE_FORMAT).timetuple())) + if date_from_uts <= bookingdate_uts <= date_to_uts: + filtered_transaction_list.append(transaction) + + self.logger.debug('transaction.Transaction._filter() ended with %s entries\n', len(filtered_transaction_list)) + return filtered_transaction_list + + def get(self, transaction_url: str, atype: str, date_from: str, date_to: str, transaction_type: str = 'booked'): + """ fetch transactions """ + self.logger.debug('transaction.Transaction.get()\n') + + if transaction_url and atype != 'depot': + transaction_url = transaction_url + '?filter[bookingDate][GE]=' + date_from + '&filter[bookingDate][LE]=' + date_to + '&expand=Merchant&page[size]=400' + + transaction_dic = self._fetch(transaction_url) + if atype == 'account': + raw_transaction_list = self._filter(transaction_list=transaction_dic['data'], transaction_type=transaction_type, date_from=date_from, date_to=date_to) + transaction = AccountTransaction(logger=self.logger) + elif atype == 'creditcard': + raw_transaction_list = self._filter(transaction_list=transaction_dic['data'], transaction_type=transaction_type, date_from=date_from, date_to=date_to) + transaction = CreditCardTransaction(logger=self.logger) + elif atype == 'depot': + transaction = DepotTransaction(logger=self.logger) + else: + raise DKBRoboError(f'transaction type {atype} is not supported') + # transaction = None + + transaction_list = [] + if transaction: + if atype != 'depot': + for ele in raw_transaction_list: + formatted_transaction = transaction.format(ele) + transaction_list.append(formatted_transaction) + else: + transaction_list = transaction.format(transaction_dic) + + self.logger.debug('transaction.Transaction.get() ended\n') + return transaction_list + + +class AccountTransaction: + """ AccountTransaction class """ + + def __init__(self, logger: logging.Logger): + self.logger = logger + + def _debitorinfo(self, transaction: Dict[str, str]) -> Dict[str, str]: + """we need debitor information for incoming payments """ + self.logger.debug('transaction.AccountTransaction._debitorinfo()\n') + + output_dic = { + 'peeraccount': transaction.get('attributes', {}).get('debtor', {}).get('debtorAccount', {}).get('iban', None), + 'peerbic': transaction.get('attributes', {}).get('debtor', {}).get('agent', {}).get('bic', None), + 'peerid': transaction.get('attributes', {}).get('debtor', {}).get('id', None), + + } + if 'attributes' in transaction and 'debtor' in transaction['attributes']: + if 'intermediaryName' in transaction['attributes']['debtor'] and transaction['attributes']['debtor']['intermediaryName']: + output_dic['peer'] = transaction.get('attributes', {}).get('debtor', {}).get('intermediaryName', None) + else: + output_dic['peer'] = transaction.get('attributes', {}).get('debtor', {}).get('name', None) + + self.logger.debug('transaction.AccountTransaction._debitorinfo() ended\n') + return output_dic + + def _creditorinfo(self, transaction: Dict[str, str]) -> Dict[str, str]: + """ we need creditor information for outgoing payments""" + self.logger.debug('transaction.AccountTransaction._creditorinfo()\n') + + output_dic = { + 'peeraccount': transaction.get('attributes', {}).get('creditor', {}).get('creditorAccount', {}).get('iban', None), + 'peerbic': transaction.get('attributes', {}).get('creditor', {}).get('agent', {}).get('bic', None), + 'peerid': transaction.get('attributes', {}).get('creditor', {}).get('id', None), + 'peer': transaction.get('attributes', {}).get('creditor', {}).get('name', None) + } + + self.logger.debug('transaction.AccountTransaction._creditorinfo() ended.\n') + return output_dic + + def _details(self, transaction: Dict[str, str]) -> Dict[str, str]: + """ add infromation from accont transaction """ + self.logger.debug('transaction.AccountTransaction._details()\n') + + try: + amount = float(transaction.get('attributes', {}).get('amount', {}).get('value', None)) + except Exception as err: + self.logger.error('amount conversion error: %s', err) + amount = None + + output_dic = { + 'amount': amount, + 'currencycode': transaction.get('attributes', {}).get('amount', {}).get('currencyCode', None), + 'date': transaction.get('attributes', {}).get('bookingDate', None), + 'vdate': transaction.get('attributes', {}).get('valueDate', None), + 'customerreference': transaction.get('attributes', {}).get('endToEndId', None), + 'mandateId': transaction.get('attributes', {}).get('mandateId', None), + 'postingtext': transaction.get('attributes', {}).get('transactionType', None), + 'reasonforpayment': transaction.get('attributes', {}).get('description', None) + } + + self.logger.debug('transaction.AccountTransaction._details() ended\n') + return output_dic + + def format(self, transaction): + """ format format transaction list ot a useful output """ + self.logger.debug('transaction.AccountTransaction.format()\n') + + if 'attributes' in transaction: + transaction_dic = self._details(transaction) + + if transaction_dic['amount'] > 0: + # incoming payment - collect debitor information + transaction_dic = {**transaction_dic, **self._debitorinfo(transaction)} + else: + # outgoing payment - collect creditor information + transaction_dic = {**transaction_dic, **self._creditorinfo(transaction)} + + # add posting text for backwards compability + if 'postingtext' in transaction_dic and 'peer' in transaction_dic and 'reasonforpayment' in transaction_dic: + transaction_dic['text'] = f'{transaction_dic["postingtext"]} {transaction_dic["peer"]} {transaction_dic["reasonforpayment"]}' + else: + transaction_dic = {} + + self.logger.debug('transaction.AccountTransaction.format() ended\n') + return transaction_dic + + +class CreditCardTransaction: + """ CreditCardTransaction class """ + + def __init__(self, logger: logging.Logger): + self.logger = logger + + def _details(self, transaction): + """ add details from card transaction """ + self.logger.debug('transaction.CreditCardTransaction._details()\n') + + if 'attributes' in transaction: + try: + amount = float(transaction.get('attributes', {}).get('amount', {}).get('value', None)) + except Exception as err: + self.logger.error('amount conversion error: %s', err) + amount = None + + output_dic = { + 'amount': amount, + 'bdate': transaction.get('attributes', {}).get('bookingDate', None), + 'vdate': transaction.get('attributes', {}).get('bookingDate', None), + 'text': transaction.get('attributes', {}).get('description', None), + 'currencycode': transaction.get('attributes', {}).get('amount', {}).get('currencyCode', None), + } + else: + output_dic = {} + + self.logger.debug('transaction.CreditCardTransaction._details() ended\n') + return output_dic + + def format(self, transaction): + """ format format transaction list ot a useful output """ + self.logger.debug('transaction.CreditCardTransaction.format()\n') + + if 'attributes' in transaction: + transaction_dic = self._details(transaction) + else: + transaction_dic = {} + + self.logger.debug('transaction.CreditCardTransaction.format() ended\n') + return transaction_dic + + +class DepotTransaction: + """ DepotTransaction class """ + + def __init__(self, logger: logging.Logger): + self.logger = logger + + def _details(self, position: Dict[str, str], included_list: List[Dict[str, str]]): + """ add details from depot transaction """ + self.logger.debug('transaction.DepotTransaction._details()\n') + + instrument_id = position.get('relationships', {}).get('instrument', {}).get('data', {}).get('id', None) + quote_id = position.get('relationships', {}).get('quote', {}).get('data', {}).get('id', None) + + try: + quantity = float(position.get('attributes', {}).get('quantity', {}).get('value', None)) + except Exception as err: + self.logger.error('quantity conversion error: %s', err) + quantity = None + + output_dic = { + 'shares': position.get('attributes', {}).get('quantity', {}).get('value', None), + 'quantity': quantity, + 'shares_unit': position.get('attributes', {}).get('quantity', {}).get('unit', None), + 'lastorderdate': position.get('attributes', {}).get('lastOrderDate', None), + 'price_euro': position.get('attributes', {}).get('performance', {}).get('currentValue', {}).get('value', None), + } + + for ele in included_list: + if 'id' in ele and ele['id'] == instrument_id: + output_dic = {**output_dic, **self._instrumentinformation(ele)} + if 'id' in ele and ele['id'] == quote_id: + output_dic = {**output_dic, **self._quoteinformation(ele)} + + self.logger.debug('transaction.DepotTransaction._details() ended\n') + return output_dic + + def _quoteinformation(self, ele: Dict[str, str]) -> Dict[str, str]: + """ add quote information """ + self.logger.debug('transaction.DepotTransaction._quoteinformation()\n') + + try: + price = float(ele.get('attributes', {}).get('price', {}).get('value', None)) + except Exception as err: + self.logger.error('price conversion error: %s', err) + price = None + + output_dic = { + 'price': price, + 'market': ele.get('attributes', {}).get('market', None), + 'currencycode': ele.get('attributes', {}).get('price', {}).get('currencyCode', None), + } + + self.logger.debug('transaction.DepotTransaction._quoteinformation() ended\n') + return output_dic + + def _instrumentinformation(self, ele: Dict[str, str]) -> Dict[str, str]: + """ add instrument information """ + self.logger.debug('transaction.DepotTransaction._instrumentinformation()\n') + + output_dic = { + 'text': ele.get('attributes', {}).get('name', {}).get('short', None) + } + if 'attributes' in ele and 'identifiers' in ele['attributes']: + for identifier in ele['attributes']['identifiers']: + if identifier['identifier'] == 'isin': + output_dic['isin_wkn'] = identifier['value'] + break + + self.logger.debug('transaction.DepotTransaction._instrumentinformation() ended\n') + return output_dic + + def format(self, transaction_dic: Dict[str, str]) -> List[Dict[str, str]]: + """ format format transaction list ot a useful output """ + self.logger.debug('transaction.DepotTransaction.format()\n') + + if 'included' in transaction_dic: + included_list = transaction_dic['included'] + + position_list = [] + if 'data' in transaction_dic: + for position in transaction_dic['data']: + position_dic = self._details(position, included_list) + if position_dic: + position_list.append(position_dic) + + return position_list diff --git a/test/test_api.py b/test/test_api.py index 8eef596..bbf788f 100644 --- a/test/test_api.py +++ b/test/test_api.py @@ -117,101 +117,6 @@ def test_008_get_loans(self): self.assertFalse(self.dkb._get_loans()) self.assertIn('ERROR:dkb_robo:api.Wrapper._get_loans(): RC is not 200 but 400', lcm.output) - @patch('dkb_robo.api.Wrapper._format_brokerage_account') - @patch('dkb_robo.api.Wrapper._format_card_transactions') - @patch('dkb_robo.api.Wrapper._format_account_transactions') - @patch('dkb_robo.api.Wrapper._filter_transactions') - def test_009_get_transactions(self, mock_ftrans, mock_atrans, mock_ctrans, mock_btrans): - """ test __legacy_get_transactions() ok """ - self.dkb.client = Mock() - self.dkb.client.get.return_value.status_code = 400 - self.dkb.client.get.return_value.json.return_value = {'foo': 'bar'} - atype = 'account' - with self.assertLogs('dkb_robo', level='INFO') as lcm: - self.assertFalse(self.dkb.get_transactions('transaction_url', atype, 'date_from', 'date_to', 'transaction_type')) - self.assertIn('ERROR:dkb_robo:api.Wrapper._get_transactions(): RC is not 200 but 400', lcm.output) - self.assertFalse(mock_atrans.called) - self.assertFalse(mock_ctrans.called) - self.assertFalse(mock_btrans.called) - - @patch('dkb_robo.api.Wrapper._format_brokerage_account') - @patch('dkb_robo.api.Wrapper._format_card_transactions') - @patch('dkb_robo.api.Wrapper._format_account_transactions') - @patch('dkb_robo.api.Wrapper._filter_transactions') - def test_010_get_transactions(self, mock_ftrans, mock_atrans, mock_ctrans, mock_btrans): - """ test __legacy_get_transactions() ok """ - self.dkb.client = Mock() - self.dkb.client.get.return_value.status_code = 200 - self.dkb.client.get.return_value.json.return_value = {} - atype = 'account' - self.assertFalse(self.dkb.get_transactions('transaction_url', atype, 'date_from', 'date_to', 'transaction_type')) - self.assertFalse(mock_atrans.called) - self.assertFalse(mock_atrans.called) - self.assertFalse(mock_atrans.called) - - @patch('dkb_robo.api.Wrapper._format_brokerage_account') - @patch('dkb_robo.api.Wrapper._format_card_transactions') - @patch('dkb_robo.api.Wrapper._format_account_transactions') - @patch('dkb_robo.api.Wrapper._filter_transactions') - def test_011_get_transactions(self, mock_ftrans, mock_atrans, mock_ctrans, mock_btrans): - """ test __legacy_get_transactions() ok """ - self.dkb.client = Mock() - self.dkb.client.get.return_value.status_code = 200 - self.dkb.client.get.return_value.json.return_value = {'foo': 'bar'} - atype = 'account' - self.assertFalse(self.dkb.get_transactions('transaction_url', atype, 'date_from', 'date_to', 'transaction_type')) - self.assertFalse(mock_atrans.called) - self.assertFalse(mock_ctrans.called) - self.assertFalse(mock_btrans.called) - - @patch('dkb_robo.api.Wrapper._format_brokerage_account') - @patch('dkb_robo.api.Wrapper._format_card_transactions') - @patch('dkb_robo.api.Wrapper._format_account_transactions') - @patch('dkb_robo.api.Wrapper._filter_transactions') - def test_012_get_transactions(self, mock_ftrans, mock_atrans, mock_ctrans, mock_btrans): - """ test __legacy_get_transactions() ok """ - self.dkb.client = Mock() - self.dkb.client.get.return_value.status_code = 200 - self.dkb.client.get.return_value.json.return_value = {'data': {'foo': 'bar'}} - atype = 'account' - mock_atrans.return_value = {'mock_foo': 'mock_bar'} - self.assertEqual({'mock_foo': 'mock_bar'}, self.dkb.get_transactions('transaction_url', atype, 'date_from', 'date_to', 'transaction_type')) - self.assertTrue(mock_atrans.called) - self.assertFalse(mock_ctrans.called) - self.assertFalse(mock_btrans.called) - - @patch('dkb_robo.api.Wrapper._format_brokerage_account') - @patch('dkb_robo.api.Wrapper._format_card_transactions') - @patch('dkb_robo.api.Wrapper._format_account_transactions') - @patch('dkb_robo.api.Wrapper._filter_transactions') - def test_013_get_transactions(self, mock_ftrans, mock_atrans, mock_ctrans, mock_btrans): - """ test __legacy_get_transactions() ok """ - self.dkb.client = Mock() - self.dkb.client.get.return_value.status_code = 200 - self.dkb.client.get.return_value.json.return_value = {'data': {'foo': 'bar'}} - atype = 'creditcard' - mock_ctrans.return_value = {'mock_foo': 'mock_bar'} - self.assertEqual({'mock_foo': 'mock_bar'}, self.dkb.get_transactions('transaction_url', atype, 'date_from', 'date_to', 'transaction_type')) - self.assertFalse(mock_atrans.called) - self.assertTrue(mock_ctrans.called) - self.assertFalse(mock_btrans.called) - - @patch('dkb_robo.api.Wrapper._format_brokerage_account') - @patch('dkb_robo.api.Wrapper._format_card_transactions') - @patch('dkb_robo.api.Wrapper._format_account_transactions') - @patch('dkb_robo.api.Wrapper._filter_transactions') - def test_014_get_transactions(self, mock_ftrans, mock_atrans, mock_ctrans, mock_btrans): - """ test __legacy_get_transactions() ok """ - self.dkb.client = Mock() - self.dkb.client.get.return_value.status_code = 200 - self.dkb.client.get.return_value.json.return_value = {'data': {'foo': 'bar'}} - atype = 'depot' - mock_btrans.return_value = {'mock_foo': 'mock_bar'} - self.assertEqual({'mock_foo': 'mock_bar'}, self.dkb.get_transactions('transaction_url', atype, 'date_from', 'date_to', 'transaction_type')) - self.assertFalse(mock_atrans.called) - self.assertFalse(mock_ctrans.called) - self.assertTrue(mock_btrans.called) - def test_015_update_token(self): """ test _update_token() ok """ self.dkb.token_dic = {'mfa_id': 'mfa_id', 'access_token': 'access_token'} @@ -813,170 +718,6 @@ def test_067__get_brokerage_details(self, mock_date): self.assertEqual(result, self.dkb._get_brokerage_details('bid', brok_dic)) self.assertFalse(mock_date.called) - def test_068__filter_transactions(self): - """ test _filter_transactions() """ - transaction_list = [] - from_date = '01.01.2023' - to_date = '31.01.2023' - self.assertFalse(self.dkb._filter_transactions(transaction_list, from_date, to_date, 'trtype')) - - def test_069__filter_transactions(self): - """ test _filter_transactions() """ - transaction_list = [{'foo': 'bar', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-15'}}] - from_date = '01.01.2023' - to_date = '31.01.2023' - result = [{'foo': 'bar', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-15'}}] - self.assertEqual(result, self.dkb._filter_transactions(transaction_list, from_date, to_date, 'trtype')) - - def test_070__filter_transactions(self): - """ test _filter_transactions() """ - transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-15'}}] - from_date = '01.01.2023' - to_date = '31.01.2023' - result = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-15'}}] - self.assertEqual(result, self.dkb._filter_transactions(transaction_list, from_date, to_date, 'trtype')) - - def test_071__filter_transactions(self): - """ test _filter_transactions() """ - transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'trtype', 'bookingDate': '2023-02-15'}}] - from_date = '01.01.2023' - to_date = '31.01.2023' - result = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}] - self.assertEqual(result, self.dkb._filter_transactions(transaction_list, from_date, to_date, 'trtype')) - - def test_072__filter_transactions(self): - """ test _filter_transactions() """ - transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'trtype2', 'bookingDate': '2023-01-15'}}] - from_date = '01.01.2023' - to_date = '31.01.2023' - result = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}] - self.assertEqual(result, self.dkb._filter_transactions(transaction_list, from_date, to_date, 'trtype')) - - def test_073__filter_transactions(self): - """ test _filter_transactions() """ - transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'trtype2', 'bookingDate': '2023-01-15'}}] - from_date = '2023-01-01' - to_date = '2023-01-31' - result = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}] - self.assertEqual(result, self.dkb._filter_transactions(transaction_list, from_date, to_date, 'trtype')) - - def test_074__filter_transactions(self): - """ test _filter_transactions() """ - transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'booked', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'pending', 'bookingDate': '2023-01-15'}}] - from_date = '01.01.2023' - to_date = '31.01.2023' - result = [{'foo1': 'bar1', 'attributes': {'status': 'booked', 'bookingDate': '2023-01-10'}}] - self.assertEqual(result, self.dkb._filter_transactions(transaction_list, from_date, to_date, 'booked')) - - def test_075__filter_transactions(self): - """ test _filter_transactions() """ - transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'booked', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'pending', 'bookingDate': '2023-01-15'}}] - from_date = '01.01.2023' - to_date = '31.01.2023' - result = [{'foo2': 'bar2', 'attributes': {'status': 'pending', 'bookingDate': '2023-01-15'}}] - self.assertEqual(result, self.dkb._filter_transactions(transaction_list, from_date, to_date, 'pending')) - - def test_076__filter_transactions(self): - """ test _filter_transactions() """ - transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'booked', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'pending', 'bookingDate': '2023-01-15'}}] - from_date = '01.01.2023' - to_date = '31.01.2023' - result = [{'foo2': 'bar2', 'attributes': {'status': 'pending', 'bookingDate': '2023-01-15'}}] - self.assertEqual(result, self.dkb._filter_transactions(transaction_list, from_date, to_date, 'reserved')) - - - def test_077_format_card_transactions(self): - """ _format_card_transactions() """ - transaction_list = [] - self.assertFalse(self.dkb._format_card_transactions(transaction_list)) - - def test_078_format_card_transactions(self): - """ _format_card_transactions() """ - transaction_list = [{'foo':'bar', 'attributes': {'description': 'description', 'bookingDate': '2023-01-01', 'amount': {'value': 1000, 'currencyCode': 'CC'}}}] - result = [{'amount': 1000.0, 'currencycode': 'CC', 'bdate': '2023-01-01', 'vdate': '2023-01-01', 'text': 'description'}] - self.assertEqual(result, self.dkb._format_card_transactions(transaction_list)) - - def test_079_format_card_transactions(self): - """ _format_card_transactions() """ - transaction_list = [{'foo':'bar', 'attributes': {'bookingDate': '2023-01-01', 'amount': {'value': 1000, 'currencyCode': 'CC'}}}] - result = [{'amount': 1000.0, 'currencycode': 'CC', 'bdate': '2023-01-01', 'vdate': '2023-01-01'}] - self.assertEqual(result, self.dkb._format_card_transactions(transaction_list)) - - def test_080_format_card_transactions(self): - """ _format_card_transactions() """ - transaction_list = [{'foo':'bar', 'attributes': {'description': 'description', 'amount': {'value': 1000, 'currencyCode': 'CC'}}}] - result = [{'amount': 1000.0, 'currencycode': 'CC', 'text': 'description'}] - self.assertEqual(result, self.dkb._format_card_transactions(transaction_list)) - - def test_081_format_brokerage_account(self): - """ test _format_brokerage_account() """ - included_list = [] - data_dic = [{'attributes': {'performance': {'currentValue': {'value': 1000}}, 'lastOrderDate': '2020-01-01', 'quantity': {'value': 1000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'id'}}, 'quote': {'data': {'id': 'id', 'value': 'value'}}}}] - brokerage_dic = {'included': included_list, 'data': data_dic} - result = [{'shares': 1000, 'quantity': 1000.0, 'shares_unit': 'unit', 'lastorderdate': '2020-01-01', 'price_euro': 1000}] - self.assertEqual(result, self.dkb._format_brokerage_account(brokerage_dic)) - - def test_082_format_brokerage_account(self): - """ test _format_brokerage_account() """ - included_list = [] - data_dic = [ - {'attributes': {'performance': {'currentValue': {'value': 1000}}, 'lastOrderDate': '2020-01-01', 'quantity': {'value': 1000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'id'}}, 'quote': {'data': {'id': 'id', 'value': 'value'}}}}, - {'attributes': {'performance': {'currentValue': {'value': 2000}}, 'lastOrderDate': '2020-02-01', 'quantity': {'value': 2000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'id2'}}, 'quote': {'data': {'id': 'id2', 'value': 'value2'}}}}] - brokerage_dic = {'included': included_list, 'data': data_dic} - result = [{'shares': 1000, 'quantity': 1000.0, 'shares_unit': 'unit', 'lastorderdate': '2020-01-01', 'price_euro': 1000}, {'shares': 2000, 'quantity': 2000.0, 'shares_unit': 'unit', 'lastorderdate': '2020-02-01', 'price_euro': 2000}] - self.assertEqual(result, self.dkb._format_brokerage_account(brokerage_dic)) - - def test_083_format_brokerage_account(self): - """ test _format_brokerage_account() """ - included_list = [{'id': 'inid', 'attributes': {'identifiers': [{'identifier': 'isin', 'value': 'value'}, {'identifier': 'isin', 'value': 'value2'}], 'name': {'short': 'short'}}}] - data_dic = [{'attributes': {'performance': {'currentValue': {'value': 1000}}, 'lastOrderDate': '2020-01-01', 'quantity': {'value': 1000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'inid'}}, 'quote': {'data': {'id': 'quoteid', 'value': 'value'}}}}] - brokerage_dic = {'included': included_list, 'data': data_dic} - result = [{'shares': 1000, 'quantity': 1000.0, 'shares_unit': 'unit', 'lastorderdate': '2020-01-01', 'price_euro': 1000, 'text': 'short', 'isin_wkn': 'value'}] - self.assertEqual(result, self.dkb._format_brokerage_account(brokerage_dic)) - - def test_084_format_brokerage_account(self): - """ test _format_brokerage_account() """ - included_list = [{'id': 'quoteid', 'attributes': {'market': 'market', 'price': {'value': 1000, 'currencyCode': 'currencyCode'}}}] - data_dic = [{'attributes': {'performance': {'currentValue': {'value': 1000}}, 'lastOrderDate': '2020-01-01', 'quantity': {'value': 1000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'inid'}}, 'quote': {'data': {'id': 'quoteid', 'value': 'value'}}}}] - brokerage_dic = {'included': included_list, 'data': data_dic} - result = [{'shares': 1000, 'quantity': 1000.0, 'shares_unit': 'unit', 'lastorderdate': '2020-01-01', 'price_euro': 1000, 'price': 1000.0, 'currencycode': 'currencyCode', 'market': 'market'}] - self.assertEqual(result, self.dkb._format_brokerage_account(brokerage_dic)) - - def test_085_format_account_transactions(self): - """ test _format_account_transactions() """ - transaction_list = [{'foo': 'bar'}] - self.assertFalse(self.dkb._format_account_transactions(transaction_list)) - - def test_086_format_account_transactions(self): - """ test _format_account_transactions() """ - transaction_list = [{'attributes': {'description': 'description', 'transactionType': 'transactionType', 'endToEndId': 'endToEndId', 'valueDate': '2023-01-02', 'bookingDate': '2023-01-01', 'debtor': {'name': 'name', 'agent': {'bic': 'bic'}, 'debtorAccount': {'iban': 'iban'}}, 'amount': {'value': 1000, 'currencyCode': 'currencyCode'}}}] - result = [{'amount': 1000.0, 'currencycode': 'currencyCode', 'peeraccount': 'iban', 'peerbic': 'bic', 'peer': 'name', 'peerid': '', 'date': '2023-01-01', 'bdate': '2023-01-01', 'vdate': '2023-01-02', 'customerreference': 'endToEndId', 'postingtext': 'transactionType', 'reasonforpayment': 'description', 'text': 'transactionType name description'}] - self.assertEqual(result, self.dkb._format_account_transactions(transaction_list)) - - def test_087_format_account_transactions(self): - """ test _format_account_transactions() """ - transaction_list = [{'attributes': {'description': 'description', 'transactionType': 'transactionType', 'endToEndId': 'endToEndId', 'valueDate': '2023-01-02', 'bookingDate': '2023-01-01', 'debtor': {'intermediaryName': 'intermediaryName', 'agent': {'bic': 'bic'}, 'debtorAccount': {'iban': 'iban'}}, 'amount': {'value': 1000, 'currencyCode': 'currencyCode'}}}] - result = [{'amount': 1000.0, 'currencycode': 'currencyCode', 'peeraccount': 'iban', 'peerbic': 'bic', 'peer': 'intermediaryName', 'peerid': '', 'date': '2023-01-01', 'bdate': '2023-01-01', 'vdate': '2023-01-02', 'customerreference': 'endToEndId', 'postingtext': 'transactionType', 'reasonforpayment': 'description', 'text': 'transactionType intermediaryName description'}] - self.assertEqual(result, self.dkb._format_account_transactions(transaction_list)) - - def test_088_format_account_transactions(self): - """ test _format_account_transactions() """ - transaction_list = [{'attributes': {'description': 'description', 'transactionType': 'transactionType', 'endToEndId': 'endToEndId', 'valueDate': '2023-01-02', 'bookingDate': '2023-01-01', 'debtor': {'id': 'id', 'name': 'name', 'agent': {'bic': 'bic'}, 'debtorAccount': {'iban': 'iban'}}, 'amount': {'value': 1000, 'currencyCode': 'currencyCode'}}}] - result = [{'amount': 1000.0, 'currencycode': 'currencyCode', 'peeraccount': 'iban', 'peerbic': 'bic', 'peer': 'name', 'peerid': 'id', 'date': '2023-01-01', 'bdate': '2023-01-01', 'vdate': '2023-01-02', 'customerreference': 'endToEndId', 'postingtext': 'transactionType', 'reasonforpayment': 'description', 'text': 'transactionType name description'}] - self.assertEqual(result, self.dkb._format_account_transactions(transaction_list)) - - def test_089_format_account_transactions(self): - """ test _format_account_transactions() """ - transaction_list = [{'attributes': {'description': 'description', 'transactionType': 'transactionType', 'endToEndId': 'endToEndId', 'valueDate': '2023-01-02', 'bookingDate': '2023-01-01', 'creditor': {'id': 'id', 'name': 'name', 'agent': {'bic': 'bic'}, 'creditorAccount': {'iban': 'iban'}}, 'amount': {'value': -1000, 'currencyCode': 'currencyCode'}}}] - result = [{'amount': -1000.0, 'currencycode': 'currencyCode', 'peeraccount': 'iban', 'peerbic': 'bic', 'peer': 'name', 'peerid': 'id', 'date': '2023-01-01', 'bdate': '2023-01-01', 'vdate': '2023-01-02', 'customerreference': 'endToEndId', 'postingtext': 'transactionType', 'reasonforpayment': 'description', 'text': 'transactionType name description'}] - self.assertEqual(result, self.dkb._format_account_transactions(transaction_list)) - - def test_090_format_account_transactions(self): - """ test _format_account_transactions() """ - transaction_list = [{'attributes': {'description': 'description', 'transactionType': 'transactionType', 'endToEndId': 'endToEndId', 'mandateId': 'mandateId', 'valueDate': '2023-01-02', 'bookingDate': '2023-01-01', 'creditor': {'name': 'name', 'agent': {'bic': 'bic'}, 'creditorAccount': {'iban': 'iban'}}, 'amount': {'value': -1000, 'currencyCode': 'currencyCode'}}}] - result = [{'amount': -1000.0, 'currencycode': 'currencyCode', 'peeraccount': 'iban', 'peerbic': 'bic', 'peer': 'name', 'mandatereference': 'mandateId', 'peerid': '', 'date': '2023-01-01', 'bdate': '2023-01-01', 'vdate': '2023-01-02', 'customerreference': 'endToEndId', 'postingtext': 'transactionType', 'reasonforpayment': 'description', 'text': 'transactionType name description'}] - self.assertEqual(result, self.dkb._format_account_transactions(transaction_list)) - @patch('dkb_robo.api.Wrapper._get_brokerage_details') @patch('dkb_robo.api.Wrapper._get_card_details') @patch('dkb_robo.api.Wrapper._get_account_details') @@ -1457,43 +1198,6 @@ def test_147_sort_mfa_devices(self): } self.assertEqual(expected_result, self.dkb._sort_mfa_devices(mfa_dic)) - def test_151_get_transaction_list(self): - """ test _get_transaction_list()""" - self.dkb.client = Mock() - self.dkb.client.get.return_value.status_code = 400 - self.dkb.client.get.return_value.json.return_value = {'foo': 'bar'} - with self.assertLogs('dkb_robo', level='INFO') as lcm: - self.assertEqual({'data': [], 'included': []}, self.dkb._get_transaction_list('transaction_url')) - self.assertIn('ERROR:dkb_robo:api.Wrapper._get_transactions(): RC is not 200 but 400', lcm.output) - - def test_152_get_transaction_list(self): - """ test _get_transaction_list()""" - self.dkb.client = Mock() - self.dkb.client.get.return_value.status_code = 200 - self.dkb.client.get.return_value.json.return_value = {'foo': 'bar'} - self.assertEqual({'data': [], 'included': []}, self.dkb._get_transaction_list('transaction_url')) - - def test_153_get_transaction_list(self): - """ test _get_transaction_list()""" - self.dkb.client = Mock() - self.dkb.client.get.return_value.status_code = 200 - self.dkb.client.get.return_value.json.return_value = {'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}]} - self.assertEqual({'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}], 'included': []}, self.dkb._get_transaction_list('transaction_url')) - - def test_154_get_transaction_list(self): - """ test _get_transaction_list()""" - self.dkb.client = Mock() - self.dkb.client.get.return_value.status_code = 200 - self.dkb.client.get.return_value.json.side_effect = [{'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}], 'links': {'next': 'next_url'}}, {'data': [{'foo3': 'bar3'}, {'foo4': 'bar4'}], 'links': {'foo': 'bar'}}] - self.assertEqual({'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}, {'foo3': 'bar3'}, {'foo4': 'bar4'}], 'included': []}, self.dkb._get_transaction_list('transaction_url')) - - def test_155_get_transaction_list(self): - """ test _get_transaction_list()""" - self.dkb.client = Mock() - self.dkb.client.get.return_value.status_code = 200 - self.dkb.client.get.return_value.json.side_effect = [{'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}], 'links': {'next': 'next_url'}, 'included': ['1']}, {'data': [{'foo3': 'bar3'}, {'foo4': 'bar4'}], 'links': {'foo': 'bar'}, 'included': ['2']}] - self.assertEqual({'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}, {'foo3': 'bar3'}, {'foo4': 'bar4'}], 'included': ['1', '2']}, self.dkb._get_transaction_list('transaction_url')) - def test_156_init(self): """ test init() """ self.dkb.__init__() diff --git a/test/test_dkb_robo.py b/test/test_dkb_robo.py index 7d359e1..7132ede 100644 --- a/test/test_dkb_robo.py +++ b/test/test_dkb_robo.py @@ -146,22 +146,24 @@ def test_008__exit(self): self.assertFalse(self.dkb.__exit__()) self.assertTrue(self.dkb.wrapper.logout.called) + @patch('dkb_robo.transaction.Transaction.get') @patch('dkb_robo.dkb_robo.validate_dates') - def test_009_get_transactions(self, mock_date): + def test_009_get_transactions(self, mock_date, mock_transaction): """ test get_transactions() """ self.dkb.legacy_login = True mock_date.return_value = ('from', 'to') self.dkb.wrapper = Mock() - self.dkb.wrapper.get_transactions.return_value = {'foo': 'bar'} - self.assertEqual({'foo': 'bar'}, self.dkb.get_transactions('url', 'atype', 'from', 'to', 'btype')) + mock_transaction.return_value = {'foo': 'bar'} + self.assertEqual({'foo': 'bar'}, self.dkb.get_transactions('url', 'account', 'from', 'to', 'btype')) + @patch('dkb_robo.transaction.Transaction.get') @patch('dkb_robo.dkb_robo.validate_dates') - def test_010_get_transactions(self, mock_date): + def test_010_get_transactions(self, mock_date, mock_transaction): """ test get_transactions() """ self.dkb.legacy_login = False mock_date.return_value = ('from', 'to') self.dkb.wrapper = Mock() - self.dkb.wrapper.get_transactions.return_value = {'foo': 'bar'} + mock_transaction.return_value = {'foo': 'bar'} self.assertEqual({'foo': 'bar'}, self.dkb.get_transactions('url', 'atype', 'from', 'to', 'btype')) def test_011_get_points(self): @@ -183,10 +185,11 @@ def test_013_get_credit_limits(self): self.dkb.wrapper.get_credit_limits.return_value = {'foo': 'bar'} self.assertEqual({'foo': 'bar'}, self.dkb.get_credit_limits()) - def test_014_get_standing_orders(self): + @patch('dkb_robo.standingorder.StandingOrder.fetch') + def test_014_get_standing_orders(self, mock_fetch): """ test get_standing_orders()""" self.dkb.wrapper = Mock() - self.dkb.wrapper.get_standing_orders.return_value = {'foo': 'bar'} + mock_fetch.return_value = {'foo': 'bar'} self.assertEqual({'foo': 'bar'}, self.dkb.get_standing_orders()) def test_015_scan_postbox(self): diff --git a/test/test_postbox.py b/test/test_postbox.py index 9db3d61..83fe853 100644 --- a/test/test_postbox.py +++ b/test/test_postbox.py @@ -264,7 +264,7 @@ def create_mocked_postbox(self, mock_client): }, ), ] - return PostBox(mock_client) + return PostBox(client=mock_client, logger=MagicMock()) @patch("requests.Session") def test_024_fetch_items(self, mock_session): @@ -299,7 +299,7 @@ def test_026_fetch_empty_responses(self, mock_session): MagicMock(status_code=200, json=lambda: {}), ] with self.assertRaises(DKBRoboError): - PostBox(mock_client).fetch_items() + PostBox(client=mock_client, logger=MagicMock()).fetch_items() @patch("requests.Session") def test_027_fetch_url_fixing(self, mock_session): @@ -317,7 +317,7 @@ def test_028_fetch_http_error(self, mock_session): mock_client = mock_session.return_value mock_client.get.side_effect = requests.HTTPError() with self.assertRaises(requests.HTTPError): - PostBox(mock_client).fetch_items() + PostBox(client=mock_client, logger=MagicMock()).fetch_items() if __name__ == "__main__": diff --git a/test/test_standingorder.py b/test/test_standingorder.py index 43a3e72..d7a03cd 100644 --- a/test/test_standingorder.py +++ b/test/test_standingorder.py @@ -42,7 +42,7 @@ def test_001_fetch(self, mock_filter): 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.assertEqual('account-id is required to fetch standing orders', str(err.exception)) self.assertFalse(mock_filter.called) @patch('dkb_robo.standingorder.StandingOrder._filter') diff --git a/test/test_transaction.py b/test/test_transaction.py new file mode 100644 index 0000000..6077570 --- /dev/null +++ b/test/test_transaction.py @@ -0,0 +1,460 @@ +# -*- 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.transaction import Transaction, AccountTransaction, CreditCardTransaction, DepotTransaction + +def json_load(fname): + """ simple json load """ + + with open(fname, 'r', encoding='utf8') as myfile: + data_dic = json.load(myfile) + + return data_dic + + +class TestTransaction(unittest.TestCase): + """ Transaction 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.transaction = Transaction(logger=self.logger, client=mock_session) + + def test_001__fetch(self): + """ test Transaction._fetch() returning error """ + self.transaction.client = Mock() + self.transaction.client.get.return_value.status_code = 400 + self.transaction.client.get.return_value.json.return_value = {'foo': 'bar'} + with self.assertLogs('dkb_robo', level='INFO') as lcm: + self.assertEqual({'data': [], 'included': []}, self.transaction._fetch('transaction_url')) + self.assertIn('ERROR:dkb_robo:fetch transactions: http status code is not 200 but 400', lcm.output) + + def test_002__fetch(self): + """ test _get_transaction_list() with wrong response""" + self.transaction.client = Mock() + self.transaction.client.get.return_value.status_code = 200 + self.transaction.client.get.return_value.json.return_value = {'foo': 'bar'} + self.assertEqual({'data': [], 'included': []}, self.transaction._fetch('transaction_url')) + + def test_003__fetch(self): + """ test _get_transaction_list() without pagination """ + self.transaction.client = Mock() + self.transaction.client.get.return_value.status_code = 200 + self.transaction.client.get.return_value.json.return_value = {'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}]} + self.assertEqual({'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}], 'included': []}, self.transaction._fetch('transaction_url')) + + def test_004__fetch(self): + """ test _get_transaction_list()""" + self.transaction.client = Mock() + self.transaction.client.get.return_value.status_code = 200 + self.transaction.client.get.return_value.json.side_effect = [{'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}], 'links': {'next': 'next_url'}}, {'data': [{'foo3': 'bar3'}, {'foo4': 'bar4'}], 'links': {'foo': 'bar'}}] + self.assertEqual({'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}, {'foo3': 'bar3'}, {'foo4': 'bar4'}], 'included': []}, self.transaction._fetch('transaction_url')) + + def test_005__fetch(self): + """ test _get_transaction_list() with pagination """ + self.transaction.client = Mock() + self.transaction.client.get.return_value.status_code = 200 + self.transaction.client.get.return_value.json.side_effect = [{'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}], 'links': {'next': 'next_url'}, 'included': ['1']}, {'data': [{'foo3': 'bar3'}, {'foo4': 'bar4'}], 'links': {'foo': 'bar'}, 'included': ['2']}] + self.assertEqual({'data': [{'foo1': 'bar1'}, {'foo2': 'bar2'}, {'foo3': 'bar3'}, {'foo4': 'bar4'}], 'included': ['1', '2']}, self.transaction._fetch('transaction_url')) + + def test_006__filter(self): + """ test _filter() with empty transaction list """ + transaction_list = [] + from_date = '01.01.2023' + to_date = '31.01.2023' + self.assertFalse(self.transaction._filter(transaction_list, from_date, to_date, 'trtype')) + + def test_007__filter(self): + """ test _filter() with a single transaction """ + transaction_list = [{'foo': 'bar', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-15'}}] + from_date = '01.01.2023' + to_date = '31.01.2023' + result = [{'foo': 'bar', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-15'}}] + self.assertEqual(result, self.transaction._filter(transaction_list, from_date, to_date, 'trtype')) + + def test_008__filter(self): + """ test _filter_transactions() with two transactions """ + transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-15'}}] + from_date = '01.01.2023' + to_date = '31.01.2023' + result = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-15'}}] + self.assertEqual(result, self.transaction._filter(transaction_list, from_date, to_date, 'trtype')) + + def test_009__filter(self): + """ test _filter_transactions() with two transactions but only one is in range """ + transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'trtype', 'bookingDate': '2023-02-15'}}] + from_date = '01.01.2023' + to_date = '31.01.2023' + result = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}] + self.assertEqual(result, self.transaction._filter(transaction_list, from_date, to_date, 'trtype')) + + def test_010__filter(self): + """ test _filter_transactions() with two transaction but only one is the right type """ + transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'trtype2', 'bookingDate': '2023-01-15'}}] + from_date = '01.01.2023' + to_date = '31.01.2023' + result = [{'foo1': 'bar1', 'attributes': {'status': 'trtype', 'bookingDate': '2023-01-10'}}] + self.assertEqual(result, self.transaction._filter(transaction_list, from_date, to_date, 'trtype')) + + def test_011__filter(self): + """ test _filter_transactions() with two transaction check for booked status """ + transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'booked', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'pending', 'bookingDate': '2023-01-15'}}] + from_date = '01.01.2023' + to_date = '31.01.2023' + result = [{'foo1': 'bar1', 'attributes': {'status': 'booked', 'bookingDate': '2023-01-10'}}] + self.assertEqual(result, self.transaction._filter(transaction_list, from_date, to_date, 'booked')) + + def test_012__filter(self): + """ test _filter_transactions() with two transactions check for pending status """ + transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'booked', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'pending', 'bookingDate': '2023-01-15'}}] + from_date = '2023-01-01' + to_date = '2023-01-31' + result = [{'foo2': 'bar2', 'attributes': {'status': 'pending', 'bookingDate': '2023-01-15'}}] + self.assertEqual(result, self.transaction._filter(transaction_list, from_date, to_date, 'pending')) + + def test_013__filter(self): + """ test _filter_transactions() with two transactions check for reserved status """ + transaction_list = [{'foo1': 'bar1', 'attributes': {'status': 'booked', 'bookingDate': '2023-01-10'}}, {'foo2': 'bar2', 'attributes': {'status': 'pending', 'bookingDate': '2023-01-15'}}] + from_date = '01.01.2023' + to_date = '31.01.2023' + result = [{'foo2': 'bar2', 'attributes': {'status': 'pending', 'bookingDate': '2023-01-15'}}] + self.assertEqual(result, self.transaction._filter(transaction_list, from_date, to_date, 'reserved')) + + @patch('dkb_robo.transaction.DepotTransaction.format') + @patch('dkb_robo.transaction.CreditCardTransaction.format') + @patch('dkb_robo.transaction.AccountTransaction.format') + @patch('dkb_robo.transaction.Transaction._filter') + @patch('dkb_robo.transaction.Transaction._fetch') + def test_014_get(self, mock_fetch, mock_filter, mock_aformat, mock_creditformat, mock_depotformat): + "" " test get() for acount transactions """ + mock_aformat.return_value = 'mock_aformat' + mock_creditformat.return_value = 'mock_creditformat' + mock_depotformat.return_value = ['mock_depotformat'] + mock_filter.return_value = ['mock_filter'] + self.assertEqual(['mock_aformat'], self.transaction.get('transaction_url', 'account', 'from_date', 'to_date')) + self.assertTrue(mock_fetch.called) + self.assertTrue(mock_filter.called) + self.assertTrue(mock_aformat.called) + self.assertFalse(mock_creditformat.called) + self.assertFalse(mock_depotformat.called) + + @patch('dkb_robo.transaction.DepotTransaction.format') + @patch('dkb_robo.transaction.CreditCardTransaction.format') + @patch('dkb_robo.transaction.AccountTransaction.format') + @patch('dkb_robo.transaction.Transaction._filter') + @patch('dkb_robo.transaction.Transaction._fetch') + def test_015_get(self, mock_fetch, mock_filter, mock_aformat, mock_creditformat, mock_depotformat): + "" " test get() for creaditcard transactions """ + mock_aformat.return_value = 'mock_aformat' + mock_creditformat.return_value = 'mock_creditformat' + mock_depotformat.return_value = ['mock_depotformat'] + mock_filter.return_value = ['mock_filter'] + self.assertEqual(['mock_creditformat'], self.transaction.get('transaction_url', 'creditcard', 'from_date', 'to_date')) + self.assertTrue(mock_fetch.called) + self.assertTrue(mock_filter.called) + self.assertFalse(mock_aformat.called) + self.assertTrue(mock_creditformat.called) + self.assertFalse(mock_depotformat.called) + + @patch('dkb_robo.transaction.DepotTransaction.format') + @patch('dkb_robo.transaction.CreditCardTransaction.format') + @patch('dkb_robo.transaction.AccountTransaction.format') + @patch('dkb_robo.transaction.Transaction._filter') + @patch('dkb_robo.transaction.Transaction._fetch') + def test_016_get(self, mock_fetch, mock_filter, mock_aformat, mock_creditformat, mock_depotformat): + "" " test get() for creaditcard transactions """ + mock_aformat.return_value = 'mock_aformat' + mock_creditformat.return_value = 'mock_creditformat' + mock_depotformat.return_value = ['mock_depotformat'] + mock_filter.return_value = ['mock_filter'] + self.assertEqual(['mock_depotformat'], self.transaction.get('transaction_url', 'depot', 'from_date', 'to_date')) + self.assertTrue(mock_fetch.called) + self.assertFalse(mock_filter.called) + self.assertFalse(mock_aformat.called) + self.assertFalse(mock_creditformat.called) + self.assertTrue(mock_depotformat.called) + + @patch('dkb_robo.transaction.DepotTransaction.format') + @patch('dkb_robo.transaction.CreditCardTransaction.format') + @patch('dkb_robo.transaction.AccountTransaction.format') + @patch('dkb_robo.transaction.Transaction._filter') + @patch('dkb_robo.transaction.Transaction._fetch') + def test_017_get(self, mock_fetch, mock_filter, mock_aformat, mock_creditformat, mock_depotformat): + "" " test get() for creaditcard transactions """ + mock_aformat.return_value = 'mock_aformat' + mock_creditformat.return_value = 'mock_creditformat' + mock_depotformat.return_value = ['mock_depotformat'] + mock_filter.return_value = ['mock_filter'] + with self.assertRaises(Exception) as err: + self.assertFalse(self.transaction.get('transaction_url', 'unknown', 'from_date', 'to_date')) + self.assertEqual('transaction type unknown is not supported', str(err.exception)) + self.assertTrue(mock_fetch.called) + self.assertFalse(mock_filter.called) + self.assertFalse(mock_aformat.called) + self.assertFalse(mock_creditformat.called) + self.assertFalse(mock_depotformat.called) + +class TestAccountTransaction(unittest.TestCase): + """ AccountTransaction test class """ + + def setUp(self): + self.dir_path = os.path.dirname(os.path.realpath(__file__)) + self.logger = logging.getLogger('dkb_robo') + self.account_transaction = AccountTransaction(logger=self.logger) + + def test_001__debitorinfo(self): + """ test _debitorinfo complete """ + transaction = {'attributes': {'debtor': {'id': 'id', 'name': 'name', 'agent': {'bic': 'bic'}, 'debtorAccount': {'iban': 'iban'}}}} + self.assertEqual({'peeraccount': 'iban', 'peerbic': 'bic', 'peerid': 'id', 'peer': 'name'}, self.account_transaction._debitorinfo(transaction)) + + def test_002__debitorinfo(self): + """ test _debitorinfo with intermediaryName and name """ + transaction = {'attributes': {'debtor': {'id': 'id', 'intermediaryName': 'intermediaryName', 'name': 'name', 'agent': {'bic': 'bic'}, 'debtorAccount': {'iban': 'iban'}}}} + self.assertEqual({'peeraccount': 'iban', 'peerbic': 'bic', 'peerid': 'id', 'peer': 'intermediaryName'}, self.account_transaction._debitorinfo(transaction)) + + def test_003__debitorinfo(self): + """ test _debitorinfo no id field""" + transaction = {'attributes': {'debtor': {'name': 'name', 'agent': {'bic': 'bic'}, 'debtorAccount': {'iban': 'iban'}}}} + self.assertEqual({'peeraccount': 'iban', 'peerbic': 'bic', 'peerid': None, 'peer': 'name'}, self.account_transaction._debitorinfo(transaction)) + + def test_004__debitorinfo(self): + """ test _debitorinfo empty id field""" + transaction = {'attributes': {'debtor': {'id': None, 'name': 'name', 'agent': {'bic': 'bic'}, 'debtorAccount': {'iban': 'iban'}}}} + self.assertEqual({'peeraccount': 'iban', 'peerbic': 'bic', 'peerid': None, 'peer': 'name'}, self.account_transaction._debitorinfo(transaction)) + + def test_005__creditorinfo(self): + """ test _creditorinfo complete """ + transaction = {'attributes': {'creditor': {'id': 'id', 'name': 'name', 'agent': {'bic': 'bic'}, 'creditorAccount': {'iban': 'iban'}}}} + self.assertEqual({'peeraccount': 'iban', 'peerbic': 'bic', 'peerid': 'id', 'peer': 'name'}, self.account_transaction._creditorinfo(transaction)) + + def test_006__creditorinfo(self): + """ test _creditorinfo complete no id field """ + transaction = {'attributes': {'creditor': {'name': 'name', 'agent': {'bic': 'bic'}, 'creditorAccount': {'iban': 'iban'}}}} + self.assertEqual({'peeraccount': 'iban', 'peerbic': 'bic', 'peerid': None, 'peer': 'name'}, self.account_transaction._creditorinfo(transaction)) + + def test_007__creditorinfo(self): + """ test _creditorinfo empty id field """ + transaction = {'attributes': {'creditor': {'id': None, 'name': 'name', 'agent': {'bic': 'bic'}, 'creditorAccount': {'iban': 'iban'}}}} + self.assertEqual({'peeraccount': 'iban', 'peerbic': 'bic', 'peerid': None, 'peer': 'name'}, self.account_transaction._creditorinfo(transaction)) + + def test_008__details(self): + """ test _details() complete """ + transaction = {'attributes': {'description': 'description', 'transactionType': 'transactionType', 'endToEndId': 'endToEndId', 'mandateId':'mandateId', 'valueDate': '2023-01-02', 'bookingDate': '2023-01-01', 'creditor': {'id': 'id', 'name': 'name', 'agent': {'bic': 'bic'}, 'creditorAccount': {'iban': 'iban'}}, 'amount': {'value': -1000, 'currencyCode': 'currencyCode'}}} + result = {'amount': -1000.0, 'currencycode': 'currencyCode', 'date': '2023-01-01', 'vdate': '2023-01-02', 'customerreference': 'endToEndId', 'mandateId': 'mandateId', 'postingtext': 'transactionType', 'reasonforpayment': 'description'} + self.assertEqual(result, self.account_transaction._details(transaction)) + + def test_009__details(self): + """ test _details() no mandateid """ + transaction = {'attributes': {'description': 'description', 'transactionType': 'transactionType', 'endToEndId': 'endToEndId', 'valueDate': '2023-01-02', 'bookingDate': '2023-01-01', 'creditor': {'id': 'id', 'name': 'name', 'agent': {'bic': 'bic'}, 'creditorAccount': {'iban': 'iban'}}, 'amount': {'value': -1000, 'currencyCode': 'currencyCode'}}} + result = {'amount': -1000.0, 'currencycode': 'currencyCode', 'date': '2023-01-01', 'vdate': '2023-01-02', 'customerreference': 'endToEndId', 'mandateId': None, 'postingtext': 'transactionType', 'reasonforpayment': 'description'} + self.assertEqual(result, self.account_transaction._details(transaction)) + + def test_010__details(self): + """ test _details() empty mandateid """ + transaction = {'attributes': {'description': 'description', 'transactionType': 'transactionType', 'endToEndId': 'endToEndId', 'mandateId': None, 'valueDate': '2023-01-02', 'bookingDate': '2023-01-01', 'creditor': {'id': 'id', 'name': 'name', 'agent': {'bic': 'bic'}, 'creditorAccount': {'iban': 'iban'}}, 'amount': {'value': -1000, 'currencyCode': 'currencyCode'}}} + result = {'amount': -1000.0, 'currencycode': 'currencyCode', 'date': '2023-01-01', 'vdate': '2023-01-02', 'customerreference': 'endToEndId', 'mandateId': None, 'postingtext': 'transactionType', 'reasonforpayment': 'description'} + self.assertEqual(result, self.account_transaction._details(transaction)) + + def test_011__details(self): + """ test _details() amount covertion error """ + transaction = {'attributes': {'description': 'description', 'transactionType': 'transactionType', 'endToEndId': 'endToEndId', 'mandateId':'mandateId', 'valueDate': '2023-01-02', 'bookingDate': '2023-01-01', 'creditor': {'id': 'id', 'name': 'name', 'agent': {'bic': 'bic'}, 'creditorAccount': {'iban': 'iban'}}, 'amount': {'value': 'aa', 'currencyCode': 'currencyCode'}}} + result = {'amount': None, 'currencycode': 'currencyCode', 'date': '2023-01-01', 'vdate': '2023-01-02', 'customerreference': 'endToEndId', 'mandateId': 'mandateId', 'postingtext': 'transactionType', 'reasonforpayment': 'description'} + with self.assertLogs('dkb_robo', level='INFO') as lcm: + self.assertEqual(result, self.account_transaction._details(transaction)) + self.assertIn("ERROR:dkb_robo:amount conversion error: could not convert string to float: 'aa'", lcm.output) + + def test_012_format(self): + """ test format() e2e for creditor transaction""" + transaction = {'attributes': {'description': 'description', 'transactionType': 'transactionType', 'endToEndId': 'endToEndId', 'valueDate': '2023-01-02', 'bookingDate': '2023-01-01', 'creditor': {'id': 'id', 'name': 'name', 'agent': {'bic': 'bic'}, 'creditorAccount': {'iban': 'iban'}}, 'amount': {'value': -1000, 'currencyCode': 'currencyCode'}}} + result = {'amount': -1000.0, 'currencycode': 'currencyCode', 'date': '2023-01-01', 'vdate': '2023-01-02', 'customerreference': 'endToEndId', 'mandateId': None, 'postingtext': 'transactionType', 'reasonforpayment': 'description', 'peeraccount': 'iban', 'peerbic': 'bic', 'peerid': 'id', 'peer': 'name', 'text': 'transactionType name description'} + self.assertEqual(result, self.account_transaction.format(transaction)) + + def test_013_format(self): + """ test format() e2e for debitor transaction""" + transaction = {'attributes': {'description': 'description', 'transactionType': 'transactionType', 'endToEndId': 'endToEndId', 'valueDate': '2023-01-02', 'bookingDate': '2023-01-01', 'debtor': {'intermediaryName': 'intermediaryName', 'agent': {'bic': 'bic'}, 'debtorAccount': {'iban': 'iban'}}, 'amount': {'value': 1000, 'currencyCode': 'currencyCode'}}} + result = {'amount': 1000.0, 'currencycode': 'currencyCode', 'date': '2023-01-01', 'vdate': '2023-01-02', 'customerreference': 'endToEndId', 'mandateId': None, 'postingtext': 'transactionType', 'reasonforpayment': 'description', 'peeraccount': 'iban', 'peerbic': 'bic', 'peerid': None, 'peer': 'intermediaryName', 'text': 'transactionType intermediaryName description'} + self.assertEqual(result, self.account_transaction.format(transaction)) + + def test_014_format(self): + """ test format() empty list """ + transaction = {'foo': 'bar'} + self.assertFalse(self.account_transaction.format(transaction)) + +class TestCreditCardTransaction(unittest.TestCase): + """ AccountTransaction test class """ + + def setUp(self): + self.dir_path = os.path.dirname(os.path.realpath(__file__)) + self.logger = logging.getLogger('dkb_robo') + self.card_transaction = CreditCardTransaction(logger=self.logger) + + def test_001_format(self): + """ test format() with empty transaction """ + transaction = {'foo': 'bar'} + self.assertFalse(self.card_transaction.format(transaction)) + + @patch('dkb_robo.transaction.CreditCardTransaction._details') + def test_002_format(self, mock_det): + """ test format() with empty transaction """ + transaction = {'attributes': {'foo': 'bar'}} + mock_det.return_value = 'mock_det' + self.assertEqual('mock_det', self.card_transaction.format(transaction)) + + def test_003__details(self): + """ test _details() with empty dictionary() """ + transaction_list = {} + self.assertFalse(self.card_transaction._details(transaction_list)) + + def test_004__details(self): + """ test _details() complete """ + transaction_list = {'foo':'bar', 'attributes': {'description': 'description', 'bookingDate': '2023-01-01', 'amount': {'value': 1000, 'currencyCode': 'CC'}}} + result = {'amount': 1000.0, 'currencycode': 'CC', 'bdate': '2023-01-01', 'vdate': '2023-01-01', 'text': 'description'} + self.assertEqual(result, self.card_transaction._details(transaction_list)) + + def test_005__details(self): + """ test _details() no description """ + transaction_list = {'foo':'bar', 'attributes': {'bookingDate': '2023-01-01', 'amount': {'value': 1000, 'currencyCode': 'CC'}}} + result = {'amount': 1000.0, 'currencycode': 'CC', 'bdate': '2023-01-01', 'vdate': '2023-01-01', 'text': None} + self.assertEqual(result, self.card_transaction._details(transaction_list)) + + def test_006__details(self): + """ test _details() complete """ + transaction_list = {'foo':'bar', 'attributes': {'description': None, 'bookingDate': '2023-01-01', 'amount': {'value': 1000, 'currencyCode': 'CC'}}} + result = {'amount': 1000.0, 'currencycode': 'CC', 'bdate': '2023-01-01', 'vdate': '2023-01-01', 'text': None} + self.assertEqual(result, self.card_transaction._details(transaction_list)) + + def test_007__details(self): + """ test _details() amount convertion error """ + transaction_list = {'foo':'bar', 'attributes': {'description': 'description', 'bookingDate': '2023-01-01', 'amount': {'value': 'aa', 'currencyCode': 'CC'}}} + result = {'amount': None, 'currencycode': 'CC', 'bdate': '2023-01-01', 'vdate': '2023-01-01', 'text': 'description'} + with self.assertLogs('dkb_robo', level='INFO') as lcm: + self.assertEqual(result, self.card_transaction._details(transaction_list)) + self.assertIn("ERROR:dkb_robo:amount conversion error: could not convert string to float: 'aa'", lcm.output) + +class TestDepotTransaction(unittest.TestCase): + """ AccountTransaction test class """ + + def setUp(self): + self.dir_path = os.path.dirname(os.path.realpath(__file__)) + self.logger = logging.getLogger('dkb_robo') + self.depot_transaction = DepotTransaction(logger=self.logger) + + @patch('dkb_robo.transaction.DepotTransaction._quoteinformation') + @patch('dkb_robo.transaction.DepotTransaction._instrumentinformation') + def test_001__details(self, mock_ii, mock_qui): + """ test _details() - add instrument information """ + mock_ii.return_value = {'mock_ii': 'mock_ii'} + mock_qui.return_value = {'mock_qui': 'mock_qui'} + included_list = [{'id': 'inid', 'attributes': {'identifiers': [{'identifier': 'isin', 'value': 'value'}, {'identifier': 'isin', 'value': 'value2'}], 'name': {'short': 'short'}}}] + data_dic = {'attributes': {'performance': {'currentValue': {'value': 1000}}, 'lastOrderDate': '2020-01-01', 'quantity': {'value': 1000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'inid'}}, 'quote': {'data': {'id': 'quoteid', 'value': 'value'}}}} + result = {'shares': 1000, 'quantity': 1000.0, 'lastorderdate': '2020-01-01', 'price_euro': 1000, 'mock_ii': 'mock_ii', 'shares_unit': 'unit'} + self.assertEqual(result, self.depot_transaction._details(data_dic, included_list)) + self.assertTrue(mock_ii.called) + self.assertFalse(mock_qui.called) + + @patch('dkb_robo.transaction.DepotTransaction._quoteinformation') + @patch('dkb_robo.transaction.DepotTransaction._instrumentinformation') + def test_002__details(self, mock_ii, mock_qui): + """ test _details() add quote information """ + mock_ii.return_value = {'mock_ii': 'mock_ii'} + mock_qui.return_value = {'mock_qui': 'mock_qui'} + included_list = [{'id': 'quoteid', 'attributes': {'identifiers': [{'identifier': 'isin', 'value': 'value'}, {'identifier': 'isin', 'value': 'value2'}], 'name': {'short': 'short'}}}] + data_dic = {'attributes': {'performance': {'currentValue': {'value': 1000}}, 'lastOrderDate': '2020-01-01', 'quantity': {'value': 1000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'inid'}}, 'quote': {'data': {'id': 'quoteid', 'value': 'value'}}}} + result = {'shares': 1000, 'quantity': 1000.0, 'lastorderdate': '2020-01-01', 'price_euro': 1000, 'mock_qui': 'mock_qui', 'shares_unit': 'unit'} + self.assertEqual(result, self.depot_transaction._details(data_dic, included_list)) + self.assertFalse(mock_ii.called) + self.assertTrue(mock_qui.called) + + @patch('dkb_robo.transaction.DepotTransaction._quoteinformation') + @patch('dkb_robo.transaction.DepotTransaction._instrumentinformation') + def test_003__details(self, mock_ii, mock_qui): + """ test _details() add instrument and quote information """ + mock_ii.return_value = {'mock_ii': 'mock_ii'} + mock_qui.return_value = {'mock_qui': 'mock_qui'} + included_list = [{'id': 'inid', 'attributes': {'identifiers': [{'identifier': 'isin', 'value': 'value'}, {'identifier': 'isin', 'value': 'value2'}], 'name': {'short': 'short'}}}] + data_dic = {'attributes': {'performance': {'currentValue': {'value': 1000}}, 'lastOrderDate': '2020-01-01', 'quantity': {'value': 1000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'inid'}}, 'quote': {'data': {'id': 'inid', 'value': 'value'}}}} + result = {'shares': 1000, 'quantity': 1000.0, 'lastorderdate': '2020-01-01', 'price_euro': 1000, 'mock_qui': 'mock_qui', 'mock_ii': 'mock_ii', 'shares_unit': 'unit'} + self.assertEqual(result, self.depot_transaction._details(data_dic, included_list)) + self.assertTrue(mock_ii.called) + self.assertTrue(mock_qui.called) + + @patch('dkb_robo.transaction.DepotTransaction._quoteinformation') + @patch('dkb_robo.transaction.DepotTransaction._instrumentinformation') + def test_004__details(self, mock_ii, mock_qui): + """ test _details() - quantity convertion error """ + mock_ii.return_value = {'mock_ii': 'mock_ii'} + mock_qui.return_value = {'mock_qui': 'mock_qui'} + included_list = [{'id': 'inid', 'attributes': {'identifiers': [{'identifier': 'isin', 'value': 'value'}, {'identifier': 'isin', 'value': 'value2'}], 'name': {'short': 'short'}}}] + data_dic = {'attributes': {'performance': {'currentValue': {'value': 1000}}, 'lastOrderDate': '2020-01-01', 'quantity': {'value': 'aa', 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'inid'}}, 'quote': {'data': {'id': 'quoteid', 'value': 'value'}}}} + result = {'shares': 'aa', 'quantity': None, 'lastorderdate': '2020-01-01', 'price_euro': 1000, 'mock_ii': 'mock_ii', 'shares_unit': 'unit'} + with self.assertLogs('dkb_robo', level='INFO') as lcm: + self.assertEqual(result, self.depot_transaction._details(data_dic, included_list)) + self.assertIn("ERROR:dkb_robo:quantity conversion error: could not convert string to float: 'aa'", lcm.output) + self.assertTrue(mock_ii.called) + self.assertFalse(mock_qui.called) + + def test_005__instrumentinformation(self): + """ test _instrumentinformation() complete """ + instrument = {'attributes': {'name': {'short': 'short'}, 'identifiers': [{'identifier': 'isin', 'value': 'value'}, {'identifier': 'isin', 'value': 'value2'}]}} + self.assertEqual({'text': 'short', 'isin_wkn': 'value'}, self.depot_transaction._instrumentinformation(instrument)) + + def test_006__instrumentinformation(self): + """ test _instrumentinformation() no identifier """ + instrument = {'attributes': {'name': {'short': 'short'}}} + self.assertEqual({'text': 'short'}, self.depot_transaction._instrumentinformation(instrument)) + + def test_007_quoteinformation(self): + """ test _quoteinformation() complete """ + quote = {'attributes': {'market': 'market', 'price': {'value': 1000, 'currencyCode': 'currencyCode'}}} + self.assertEqual({'currencycode': 'currencyCode', 'market': 'market', 'price': 1000.0}, self.depot_transaction._quoteinformation(quote)) + + def test_008_quoteinformation(self): + """ test _quoteinformation() error value convert """ + quote = {'attributes': {'market': 'market', 'price': {'value': 'aaa', 'currencyCode': 'currencyCode'}}} + with self.assertLogs('dkb_robo', level='INFO') as lcm: + self.assertEqual({'currencycode': 'currencyCode', 'market': 'market', 'price': None}, self.depot_transaction._quoteinformation(quote)) + self.assertIn("ERROR:dkb_robo:price conversion error: could not convert string to float: 'aaa'", lcm.output) + + def test_009_format(self): + """ test _format_brokerage_account() """ + included_list = [] + data_dic = [{'attributes': {'performance': {'currentValue': {'value': 1000}}, 'lastOrderDate': '2020-01-01', 'quantity': {'value': 1000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'id'}}, 'quote': {'data': {'id': 'id', 'value': 'value'}}}}] + brokerage_dic = {'included': included_list, 'data': data_dic} + result = [{'shares': 1000, 'quantity': 1000.0, 'shares_unit': 'unit', 'lastorderdate': '2020-01-01', 'price_euro': 1000}] + self.assertEqual(result, self.depot_transaction.format(brokerage_dic)) + + def test_010_format(self): + """ test format() e2e """ + included_list = [] + data_dic = [ + {'attributes': {'performance': {'currentValue': {'value': 1000}}, 'lastOrderDate': '2020-01-01', 'quantity': {'value': 1000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'id'}}, 'quote': {'data': {'id': 'id', 'value': 'value'}}}}, + {'attributes': {'performance': {'currentValue': {'value': 2000}}, 'lastOrderDate': '2020-02-01', 'quantity': {'value': 2000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'id2'}}, 'quote': {'data': {'id': 'id2', 'value': 'value2'}}}}] + brokerage_dic = {'included': included_list, 'data': data_dic} + result = [{'shares': 1000, 'quantity': 1000.0, 'shares_unit': 'unit', 'lastorderdate': '2020-01-01', 'price_euro': 1000}, {'shares': 2000, 'quantity': 2000.0, 'shares_unit': 'unit', 'lastorderdate': '2020-02-01', 'price_euro': 2000}] + self.assertEqual(result, self.depot_transaction.format(brokerage_dic)) + + def test_011_format(self): + """ test format() e2e """ + included_list = [{'id': 'inid', 'attributes': {'identifiers': [{'identifier': 'isin', 'value': 'value'}, {'identifier': 'isin', 'value': 'value2'}], 'name': {'short': 'short'}}}] + data_dic = [{'attributes': {'performance': {'currentValue': {'value': 1000}}, 'lastOrderDate': '2020-01-01', 'quantity': {'value': 1000, 'unit': 'unit'}}, 'relationships': {'instrument': {'data': {'id': 'inid'}}, 'quote': {'data': {'id': 'quoteid', 'value': 'value'}}}}] + brokerage_dic = {'included': included_list, 'data': data_dic} + result = [{'shares': 1000, 'quantity': 1000.0, 'shares_unit': 'unit', 'lastorderdate': '2020-01-01', 'price_euro': 1000, 'text': 'short', 'isin_wkn': 'value'}] + self.assertEqual(result, self.depot_transaction.format(brokerage_dic)) + + +if __name__ == '__main__': + + unittest.main()