diff --git a/.gitignore b/.gitignore index 2efa43d..e34e2e2 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ +.vscode/settings.json + *.py[cod] # C extensions @@ -37,3 +39,6 @@ htmlcov/* .pydevproject .eggs/ .pytest_cache/ + +# env variables +.env diff --git a/.vscode/settings.json b/.vscode/settings.json deleted file mode 100644 index d593deb..0000000 --- a/.vscode/settings.json +++ /dev/null @@ -1,4 +0,0 @@ -{ - "python.pythonPath": "/usr/local/bin/python", - "python.linting.pylintEnabled": true -} \ No newline at end of file diff --git a/figo/credentials.py b/figo/credentials.py index 9ca2e86..f4c3983 100644 --- a/figo/credentials.py +++ b/figo/credentials.py @@ -1,16 +1,4 @@ import os -DEMO_CREDENTIALS = { - 'client_id': 'C-9rtYgOP3mjHhw0qu6Tx9fgk9JfZGmbMqn-rnDZnZwI', - 'client_secret': 'Sv9-vNfocFiTe_NoMRkvNLe_jRRFeESHo8A0Uhyp7e28', - 'api_endpoint': 'https://api.figo.me', -} - DEMO_TOKEN = ('ASHWLIkouP2O6_bgA2wWReRhletgWKHYjLqDaqb0LFfamim9RjexTo' - '22ujRIP_cjLiRiSyQXyt2kM1eXU2XLFZQ0Hro15HikJQT_eNeT_9XQ') - -CREDENTIALS = { - 'client_id': os.getenv('FIGO_CLIENT_ID', DEMO_CREDENTIALS['client_id']), - 'client_secret': os.getenv('FIGO_CLIENT_SECRET', DEMO_CREDENTIALS['client_secret']), - 'api_endpoint': os.getenv('FIGO_API_ENDPOINT', DEMO_CREDENTIALS['api_endpoint']), -} + '22ujRIP_cjLiRiSyQXyt2kM1eXU2XLFZQ0Hro15HikJQT_eNeT_9XQ') \ No newline at end of file diff --git a/figo/figo.py b/figo/figo.py index 7ecf629..4d1c249 100644 --- a/figo/figo.py +++ b/figo/figo.py @@ -9,6 +9,11 @@ import logging import re import sys +import urllib +import os + +from dotenv import load_dotenv +load_dotenv() from datetime import datetime from datetime import timedelta @@ -16,21 +21,24 @@ from requests import Session from time import sleep -from figo.credentials import CREDENTIALS +# from figo.credentials import CREDENTIALS from figo.models import Account from figo.models import AccountBalance from figo.models import BankContact +from figo.models import Challenge from figo.models import LoginSettings from figo.models import Notification from figo.models import Payment from figo.models import PaymentProposal from figo.models import Security from figo.models import Service +from figo.models import StandingOrder from figo.models import TaskState from figo.models import TaskToken from figo.models import Transaction from figo.models import User from figo.models import WebhookNotification +from figo.models import Sync from figo.version import __version__ @@ -41,6 +49,7 @@ logger = logging.getLogger(__name__) +API_ENDPOINT = os.getenv("API_ENDPOINT") ERROR_MESSAGES = { 400: { @@ -75,12 +84,29 @@ }, } +def getAccountId(account_or_account_id): + if account_or_account_id == None: + return None + elif isinstance(account_or_account_id, Account): + return account_or_account_id.account_id + else: + return account_or_account_id + +def filterKeys(object, allowed_keys): + if object == None or object == {}: + return {} + else: + keys = [key for key in object.keys() if key in allowed_keys] + return dict(zip(keys, [object[key] for key in keys])) + +def filterNone(object): + return { k: v for k, v in object.items() if v is not None } class FigoObject(object): """A FigoObject has the ability to communicate with the Figo API.""" def __init__(self, - api_endpoint=CREDENTIALS['api_endpoint'], + api_endpoint=API_ENDPOINT, language=None): """Create a FigoObject instance. @@ -107,7 +133,6 @@ def _request_api(self, path, data=None, method="GET"): Returns: the JSON-parsed result body """ - complete_path = self.api_endpoint + path session = Session() @@ -155,6 +180,9 @@ def _query_api_object(self, model, path, data=None, method="GET", collection_nam return None elif collection_name is None: return model.from_dict(self, response) + elif collection_name == "collection": + # Some collections in the API response ARE NOT embbeded in collection name. (Ex: challenges, accesses) + return [model.from_dict(self, dict_entry) for dict_entry in response] else: return [model.from_dict(self, dict_entry) for dict_entry in response[collection_name]] @@ -230,7 +258,7 @@ class FigoConnection(FigoObject): """ def __init__(self, client_id, client_secret, redirect_uri, - api_endpoint=CREDENTIALS['api_endpoint'], + api_endpoint=API_ENDPOINT, language=None): """ Create a FigoConnection instance. @@ -339,11 +367,10 @@ def credential_login(self, username, password, scope=None): Dictionary which contains an access token and a refresh token. """ - data = {"grant_type": "password", + data = filterNone({"grant_type": "password", "username": username, - "password": password} - if scope: - data["scope"] = scope + "password": password, + "scope": scope}) response = self._request_api("/auth/token", data, method="POST") @@ -412,11 +439,10 @@ def add_user(self, name, email, password, language='de'): """ response = self._request_api( path="/auth/user", - data={'name': name, + data={'full_name': name, 'email': email, 'password': password, - 'language': language, - 'affiliate_client_id': self.client_id}, + 'language': language}, method="POST") if response is None: @@ -424,7 +450,7 @@ def add_user(self, name, email, password, language='de'): elif 'error' in response: raise FigoException.from_dict(response) else: - return response['recovery_password'] + return response def add_user_and_login(self, name, email, password, language='de'): """ @@ -444,13 +470,37 @@ def add_user_and_login(self, name, email, password, language='de'): self.add_user(name, email, password, language) return self.credential_login(email, password) + def get_version(self): + """ + Returns the version of the API. + """ + return self._request_api(path="/version", method='GET') + + def get_catalog(self, q=None, country_code=None): + """Return a dict with lists of supported banks and payment services, with client auth. + + Returns: + dict {'banks': [BankContact], 'services': [Service]}: + dict with lists of supported banks and payment services + """ + options = filterNone({ "country": country_code, "q": q }) + catalog = self._query_api("/catalog?" + urllib.urlencode(options)) + + for k, v in catalog.items(): + if k == 'banks': + catalog[k] = [BankContact.from_dict(self, bank) for bank in v] + elif k == 'services': + catalog[k] = [Service.from_dict(self, service) for service in v] + + return catalog + class FigoSession(FigoObject): """ Represents a user-bound connection to the figo connect API and allows access to the users data. """ def __init__(self, access_token, sync_poll_retry=20, - api_endpoint=CREDENTIALS['api_endpoint'], + api_endpoint=API_ENDPOINT, language=None, ): """Create a FigoSession instance. @@ -474,161 +524,115 @@ def accounts(self): """ return self._query_api_object(Account, "/rest/accounts", collection_name="accounts") - def get_account(self, account_id): - """Retrieve a specific account. + def get_account(self, account_or_account_id, cents=None): + """ + Args: + account_or_account_id: account to be queried or its ID + cents (bool): If true amounts will be shown in cents, Optional, default: False - Args: - account_id: id of the account to be retrieved + Returns: + Account: An account accessible from Token + """ + options = { "cents": cents } if cents else {} + return self._query_api_object(Account, path="/rest/accounts/{0}".format(getAccountId(account_or_account_id)), method='GET') - Returns: - Account object for the respective account - """ - return self._query_api_object(Account, "/rest/accounts/%s" % account_id) + def get_accounts(self): + """ + Args: + None + + Returns: + List of Accounts accessible from Token + """ - def add_account(self, country, credentials, bank_code=None, iban=None, save_pin=False): - """Add a bank account to the figo user. + return self._query_api_object(Account, path="/rest/accounts", method='GET', collection_name="accounts") + + def modify_account(self, account): + """Modify an account. Args: - country (str): country code of the bank to add - credentials ([str]): list of credentials needed for bank login - bank_code (str): bank code of the bank to add - iban (str): iban of the account to add - save_pin (bool): save credentials on the figo Connect server + account: the modified account to be saved Returns: - TaskToken: A task token for the account creation task - - Note: - `bank_code` or `iban` must be set, and `iban` overrides `bank_code`. + Account object for the updated account returned by server """ - data = {'country': country, 'credentials': credentials, 'save_pin': save_pin} - if iban: - data['iban'] = iban - elif bank_code: - data['bank_code'] = bank_code + return self._query_api_object(Account, "/rest/accounts/%s" % account.account_id, + account.dump(), "PUT") - return self._query_api_object(TaskToken, "/rest/accounts", data, "POST") + def remove_account(self, account_or_account_id): + """Remove an account. - def add_account_and_sync(self, country, credentials, bank_code=None, iban=None, save_pin=False): - """Add a bank account and start syncing it. + Args: + account_or_account_id: account to be removed or its ID + """ + path = "/rest/accounts/{0}".format(getAccountId(account_or_account_id)) + self._request_with_exception(path, method="DELETE") + def add_sync(self, access_id, disable_notifications, redirect_uri, state, credentials, save_secrets): + """ Args: - country (str): country code of the bank to add - credentials ([str]): list of credentials needed for bank login - bank_code (str): bank code of the bank to add - iban (str): iban of the account to add - save_pin (bool): save credentials on the figo Connect server + access_id (str): figo ID of the provider access, Required + disable_notifications (bool): This flag indicates whether notifications should be sent to your + application, Optional, default: False + redirect_uri (str): The URI to which the end user is redirected in OAuth cases, Optional + state (str): Arbitrary string to maintain state between this request and the callback + credentials (obj): Credentials used for authentication with the financial service provider. + save_secrets (bool): Indicates whether the confidential parts of the credentials should be saved, default: False Returns: - TaskToken: A task token for the account creation task - - Note: - `bank_code` or `iban` must be set, and `iban` overrides `bank_code`. - The number of sync retries is determined by `FigoSession.sync_poll_retry`. - """ - task_token = self.add_account(country, credentials, bank_code, iban, save_pin) - for _ in range(self.sync_poll_retry): - task_state = self.get_task_state(task_token) - logger.info("Adding account {0}/{1}: {2}".format(bank_code, iban, task_state.message)) - logger.debug(str(task_state)) - if task_state.is_ended or task_state.is_erroneous: - break - sleep(2) - else: - raise FigoException( - "could not sync", - "task was not finished after {0} tries".format(self.sync_poll_retry) - ) - - if task_state.is_erroneous: - if task_state.error and task_state.error['code'] == 10000: - raise FigoPinException(country, credentials, bank_code, iban, save_pin, - error=task_state.error['name'], - error_description=task_state.error['description'], - code=task_state.error['code']) - raise FigoException("", error_description=task_state.error['message'], - code=task_state.error['code']) - return task_state - - def add_account_and_sync_with_new_pin(self, pin_exception, new_pin): - """Provide a new pin if the sync task was erroneous because of a wrong pin. + Object: synchronization operation. + """ + data = filterNone({ "disable_notifications": disable_notifications, "redirect_uri": redirect_uri, + "state": state, "credentials": credentials, "save_secrets": save_secrets}) + return self._query_api_object(Sync, "/rest/accesses/{0}/syncs".format(access_id), data=data, method='POST') + + def get_synchronization_status(self, access_id, sync_id): + """ Args: - pin_exception: Exception of the sync task for which a new pin will be provided - new_pin: New pin for the sync task + access_id (str): figo ID of the provider access, Required + sync_id (str): figo ID of the synchronization operation, Required Returns: - The state of the sync task. If the pin was wrong a FigoPinException is thrown - """ - pin_exception.credentials[1] = new_pin - return self.add_account_and_sync( - pin_exception.country, - pin_exception.credentials, - pin_exception.bank_code, - pin_exception.iban, - pin_exception.save_pin, - ) - - def modify_account(self, account): - """Modify an account. + Object: synchronization operation. + """ + return self._query_api_object(Sync, "/rest/accesses/{0}/syncs/{1}".format(access_id, sync_id), method='GET') + def get_synchronization_challenges(self, access_id, sync_id): + """ Args: - account: the modified account to be saved + access_id (str): figo ID of the provider access, Required + sync_id (str): figo ID of the synchronization operation, Required Returns: - Account object for the updated account returned by server + Object: List of challenges associated with synchronization operation. """ - return self._query_api_object(Account, "/rest/accounts/%s" % account.account_id, - account.dump(), "PUT") - - def remove_account(self, account_or_account_id): - """Remove an account. + return self._query_api_object(Challenge, "/rest/accesses/{0}/syncs/{1}/challenges".format(access_id, sync_id), method='GET', collection_name="collection") - Args: - account_or_account_id: account to be removed or its ID + def get_synchronization_challenge(self, access_id, sync_id, challenge_id): """ - if isinstance(account_or_account_id, Account): - account_or_account_id = account_or_account_id.account_id + Args: + access_id (str): figo ID of the provider access, Required + sync_id (str): figo ID of the synchronization operation, Required + challenge_id (str): figo ID of the challenge, Required - query = "/rest/accounts/{0}".format(account_or_account_id) - self._request_with_exception(query, method="DELETE") + Returns: + Object: Challenge associated with synchronization operation. + """ + return self._query_api_object(Challenge, "/rest/accesses/{0}/syncs/{1}/challenges/{2}".format(access_id, sync_id, challenge_id), method='GET') - def sync_account(self, state, redirect_uri=None, account_ids=None, if_not_synced_since=None, - sync_tasks=['transactions'], disable_notifications=False, auto_continue=False): + def solve_synchronization_challenge(self, access_id, sync_id, challenge_id, data): """ Args: - state (str): Arbitrary string to maintain state between this request and the callback, - e.g. it might contain a session ID from your application. - The value should also contain a random component, which your - application checks to prevent cross-site request forgery. - redirect_uri (str): At the end of the synchronization process a response will be sent to - this callback URL. The value defaults to the first redirect URI - configured for the client. - disable_notifications (bool): This flag indicates whether notifications should be sent - to your application. Since your application will be notified by - the callback URL anyway, you might want to disable any - additional notifications. - if_not_synced_since (int): If this parameter is set, only those accounts will be - synchronized, which have not been synchronized within the - specified number of minutes. - auto_continue (bool): Automatically acknowledge and ignore any errors. - account_ids ([str]): Only sync the accounts with these IDs. + access_id (str): figo ID of the provider access, Required + sync_id (str): figo ID of the synchronization operation, Required + challenge_id (str): figo ID of the challenge, Required Returns: - TaskToken: A task token for the synchronization task + {} """ - data = { - 'state': state, - 'redirect_uri': redirect_uri, - 'disable_notifications': disable_notifications, - 'if_not_synced_since': if_not_synced_since, - 'auto_continue': auto_continue, - 'account_ids': account_ids, - 'sync_tasks': sync_tasks, - } - - data = dict((k, v) for k, v in data.items() if v is not None) # noqa, py26 compatibility - return self._query_api_object(model=TaskToken, path='/rest/sync', data=data, method='POST') + return self._request_api(path="/rest/accesses/{0}/syncs/{1}/challenges/{2}/response" + .format(access_id, sync_id, challenge_id), data=data, method='POST') def get_account_balance(self, account_or_account_id): """Get balance and account limits. @@ -661,19 +665,68 @@ def modify_account_balance(self, account_or_account_id, account_balance): query = "/rest/accounts/{0}/balance".format(account_or_account_id) return self._query_api_object(AccountBalance, query, account_balance.dump(), "PUT") - def get_catalog(self): + def get_catalog(self, country_code=None): """Return a dict with lists of supported banks and payment services. Returns: dict {'banks': [Service], 'services': [Service]}: dict with lists of supported banks and payment services """ - catalog = self._request_with_exception("/rest/catalog") + options = filterNone({ "country": country_code }) + + catalog = self._request_with_exception("/rest/catalog?" + urllib.urlencode(options)) for k, v in catalog.items(): catalog[k] = [Service.from_dict(self, service) for service in v] return catalog + def add_access(self, access_method_id, credentials, consent): + """Add provider access + + Args: + access_method_id (str): figo ID of the provider access method. [required] + credentials (Crendentials object): Credentials used for authentication with the financial service provider. + consent (Consent object): Configuration of the PSD2 consents. Is ignored for non-PSD2 accesses. + + Returns: + Access object added + """ + data = { "access_method_id": access_method_id, "credentials" : credentials, "consent": consent } + return self._request_api(path="/rest/accesses", data=data, method="POST") + + def get_accesses(self): + """List all connected provider accesses of user. + + Returns: + Array of Access objects + """ + return self._request_with_exception("/rest/accesses") + + def get_access(self, access_id): + """Retrieve the details of a specific provider access identified by its ID. + + Args: + access_id (str): figo ID of the provider access. [required] + + Returns: + Access object matching the access_id + """ + return self._request_with_exception("/rest/accesses/{0}".format(access_id), method="GET") + + def remove_pin(self, access_id): + """Remove a PIN from the API backend that has been previously stored for automatic synchronization or ease of use. + + Args: + access_id (str): figo ID of the provider access. [required] + + Returns: + Access object for which the PIN was removed + """ + return self._request_api( + path="/rest/accesses/%s/remove_pin" % access_id, + method="POST" + ) + def get_supported_payment_services(self, country_code): """Return a list of supported credit cards and other payment services. @@ -728,6 +781,31 @@ def get_service_login_settings(self, country_code, item_id): return self._query_api_object(LoginSettings, "/rest/catalog/services/%s/%s" % (country_code, item_id)) + def get_standing_orders(self, account_or_account_id=None, accounts=None, count=None, offset=None, cents=None): + """Get an array of `StandingOrder` objects, one for each standing order of the user on + the specified account. + + Args: + account_or_account_id: account to be queried or its ID, Optional + Accounts: Comma separated list of account IDs, Optional + Count: Limit the number of returned items, Optional + Offset: Skip this number of transactions in the response, Optional + Cents: If true amounts will be shown in cents, Optional, default: False + + Returns: + List of standing order objects + """ + + options = filterNone({ "accounts": accounts, "count": count, "offset": offset, "cents": cents }) + + account_id = getAccountId(account_or_account_id) + if account_id: + query = "/rest/accounts/{0}/standing_orders?{1}".format(account_id, urllib.urlencode(options)) + else: + query = "/rest/standing_orders?{0}".format(urllib.urlencode(options)) + + return self._query_api_object(StandingOrder, query, collection_name="standing_orders") + @property def notifications(self): """An array of `Notification` objects, one for each registered notification.""" @@ -791,36 +869,45 @@ def payments(self): """ return self._query_api_object(Payment, "/rest/payments", collection_name="payments") - def get_payments(self, account_or_account_id): + def get_payments(self, account_or_account_id, accounts, count, offset, cents): """Get an array of `Payment` objects, one for each payment of the user on the specified account. Args: account_or_account_id: account to be queried or its ID + Accounts: Comma separated list of account IDs. + Count: Limit the number of returned items, Optional + Offset: Skip this number of transactions in the response, Optional + Cents: If true amounts will be shown in cents, Optional, default: False Returns: List of Payment objects """ - if isinstance(account_or_account_id, Account): - account_or_account_id = account_or_account_id.account_id - query = "/rest/accounts/{0}/payments".format(account_or_account_id) + options = filterNone({ "accounts": accounts, "count": count, "offset": offset, "cents": cents }) + + account_id = getAccountId(account_or_account_id) + if account_id: + query = "/rest/accounts/{0}/payments?{1}".format(account_id, urllib.urlencode(options)) + else: + query = "/rest/payments?{0}".format(urllib.urlencode(options)) + return self._query_api_object(Payment, query, collection_name="payments") - def get_payment(self, account_or_account_id, payment_id): + def get_payment(self, account_or_account_id, payment_id, cents): """Get a single `Payment` object. Args: account_or_account_id: account to be queried or its ID payment_id: ID of the payment to be retrieved + Cents (bool): If true amounts will be shown in cents, Optional, default: False Returns: Payment object """ - if isinstance(account_or_account_id, Account): - account_or_account_id = account_or_account_id.account_id + options = { "cents": cents } if cents else {} - query = "/rest/accounts/{0}/payments/{1}".format(account_or_account_id, payment_id) + query = "/rest/accounts/{0}/payments/{1}?{2}".format(getAccountId(account_or_account_id), payment_id, urllib.urlencode(options)) return self._query_api_object(Payment, query) def add_payment(self, payment): @@ -832,9 +919,7 @@ def add_payment(self, payment): Returns: Payment object of the newly created payment as returned by the server """ - return self._query_api_object(Payment, - "/rest/accounts/{0}/payments".format(payment.account_id), - payment.dump(), "POST") + return self._query_api_object(Payment, "/rest/accounts/{0}/payments".format(payment.account_id), payment.dump(), "POST") def modify_payment(self, payment): """Modify a payment. @@ -859,31 +944,129 @@ def remove_payment(self, payment): method="DELETE") def submit_payment(self, payment, tan_scheme_id, state, redirect_uri=None): - """Submit payment to bank server. + """Submit payment to bank server. + + Args: + payment: payment to be submitted, Required + tan_scheme_id: TAN scheme ID of user-selected TAN scheme, Required + state: Any kind of string that will be forwarded in the callback response message, Required + redirect_uri: At the end of the submission process a response will + be sent to this callback URL, Optional + + Returns: + the URL to be opened by the user for the TAN process + """ + params = {'tan_scheme_id': tan_scheme_id, 'state': state} + if redirect_uri is not None: + params['redirect_uri'] = redirect_uri + + response = self._request_with_exception( + "/rest/accounts/%s/payments/%s/init" % (payment.account_id, payment.payment_id), + params, "POST") + + def get_payment_status(self, payment_id, init_id): + """Get initiation status for payment initiated to bank server. + + Args: + payment: payment to be retrieved the status for, Required + init_id: initiation id, Required + + Returns: + the initiation status of the payment + """ + response = self._request_with_exception( + "/rest/accounts/%s/payments/%s/init/%s" % (payment.account_id, payment.payment_id, init_id), None, "GET") + + def get_payment_challenges(self, account_or_account_id, payment_id, init_id): + """List payment challenges + + Args: + account_or_account_id: account to be queried or its ID, Required + payment: payment to be retrieved the status for, Required + init_id: initiation id, Required + + Returns: + List of challenges for the required payment + """ + account_id = getAccountId(account_or_account_id) + + return self._query_api_object(Challenge, "/rest/accounts/{0}/payments/{1}/init/{2}/challenges".format(account_id, payment_id, init_id), "GET") + + def get_payment_challenge(self, account_or_account_id, payment_id, init_id, challenge_id): + """Get payment challenge + + Args: + account_or_account_id: account to be queried or its ID, Required + payment: payment to be retrieved the status for, Required + init_id: initiation id, Required + challenge_id: challenge id, Required + + Returns: + Challenge: The required challenge for the payment + """ + account_id = getAccountId(account_or_account_id) + + return self._query_api_object(Challenge, "/rest/accounts/{0}/payments/{1}/init/{2}/challenges/{3}".format(account_id, payment_id, init_id, challenge_id), "GET") + + def solve_payment_challenges(self, account_or_account_id, payment_id, init_id, challenge_id, payload): + """Get payment challenge + + Args: + account_or_account_id (str): account to be queried or its ID, Required + payment (str): payment to be retrieved the status for, Required + init_id (str): initiation id, Required + challenge_id (str): challenge id, Required + payload (one of): + AuthMethodSelectResponse: + - method_id (str): figo ID of TAN scheme. + ChallengeResponse: + value (str): Response to the auth challenge. The source of the value depends on the selected authentication method. + ChallengeResponseJWE: + - type (str): The type of the value. Always set to the value "encrypted". + - value (str): JWE encrypted auth challenge response. + + Returns: + Challenge: The required challenge for the payment + """ + account_id = getAccountId(account_or_account_id) + + return self._query_api_object(Challenge, "/rest/accounts/{0}/payments/{1}/init/{2}/challenges/{3}/response".format(account_id, payment_id, init_id, challenge_id), payload, "POST") + + @property + def get_standing_order(self, standing_order_id, account_or_account_id=None, cents=None): + """Get a single `StandingOrder` object. Args: - payment: payment to be submitted - tan_scheme_id: TAN scheme ID of user-selected TAN scheme - state: Any kind of string that will be forwarded in the callback response message - redirect_uri: At the end of the submission process a response will - be sent to this callback URL + standing_order_id: ID of the standing order to be retrieved, Required + account_or_account_id: account to be queried or its ID, Optional + Cents (bool): If true amounts will be shown in cents, Optional, default: False Returns: - the URL to be opened by the user for the TAN process + standing order object """ - params = {'tan_scheme_id': tan_scheme_id, 'state': state} - if redirect_uri is not None: - params['redirect_uri'] = redirect_uri + options = filterNone({ "accounts": accounts, "cents": cents }) - response = self._request_with_exception( - "/rest/accounts/%s/payments/%s/submit" % (payment.account_id, payment.payment_id), - params, "POST") + account_id = getAccountId(account_or_account_id) + if account_id: + query = "/rest/accounts/{0}/standing_orders/{1}?{2}".format(account_id, standing_order_id, urllib.urlencode(options)) + else: + query = "/rest/standing_orders/{0}?{1}".format(standing_order_i, urllib.urlencode(options)) - if response is None: - return None + return self._query_api_object(StandingOrder, query) + + def remove_standing_order(self, StandingOrder, account_or_account_id=None): + """Remove a standing order. + + Args: + StandingOrder: standing order to be removed, Required + account_or_account_id: account to be queried or its ID, Optional + """ + if account_id: + path = "/rest/accounts/{0}/standing_orders/{1}".format(account_id, standing_order_id) else: - return (self.api_endpoint + "/task/start?id=" + - response["task_token"]) + path = "/rest/standing_orders/{0}".format(standing_order_id) + + self._request_with_exception(path, method="DELETE") @property def payment_proposals(self): @@ -959,63 +1142,81 @@ def transactions(self): return self._query_api_object(Transaction, "/rest/transactions", collection_name="transactions") - def get_transactions(self, account_id=None, since=None, count=1000, offset=0, - include_pending=False, sort='desc'): - """Get an array of `Transaction` objects, one for each transaction of the user. + def get_transactions(self, account_or_account_id, options ): + """Get an array of Transaction, one for each transaction of the user. Args: - account_id (str): ID of the account for which to list the transactions - since (str): This parameter can either be a transaction ID or a date. + account_or_account_id (str): ID of the account for which to list the transactions OR account object. + + options (obj): further optional options + accounts: comma separated list of account IDs. + filter (obj) - Can take 4 possible keys: + - date (ISO date) - Transaction date + - person (str) - Payer or payee name + - purpose (str) + - amount (num) + sync_id (str): Show only those items that have been created within this synchronization. count (int): Limit the number of returned transactions. offset (int): Which offset into the result set should be used to determine the - first transaction to return (useful in combination with count) + first transaction to return (useful in combination with count) + sort (enum): ASC or DESC + since (ISO date): Return only transactions after this date based on since_type + until (ISO date): This parameter can either be a transaction ID or a date. Return only transactions which were booked on or before + since_type (enum): This parameter defines how the parameter since will be interpreted. + Possible values: "booked" "created" "modified" + types (enum): Comma separated list of transaction types used for filtering. + Possible values:"Transfer", "Standing order", "Direct debit", "Salary or rent", "GeldKarte", "Charges or interest" + cents (bool): If true amounts will be shown in cents, Optional, default: False include_pending (bool): This flag indicates whether pending transactions should - be included in the response. Pending transactions are always - included as a complete set, regardless of the `since` parameter. + be included in the response. Pending transactions are always + included as a complete set, regardless of the `since` parameter. + include_statistics (bool): Includes statistics on the returned transactionsif true, Default: false. Returns: - [Transaction]: List of `Transaction` objects + List of Transaction """ - params = {'count': count, 'offset': offset, 'sort': sort, - 'include_pending': ("1" if include_pending else "0")} - if since is not None: - params['since'] = since - - params = urllib.urlencode(params) + allowed_keys = ["accounts", "filter", "sync_id", "count", "offset", "sort", "since", "until", "since_type", "types", "cents", "include_pending", "include_statistics"] + options = filterNone(filterKeys(options, allowed_keys)) + account_id = getAccountId(account_or_account_id) if account_id is not None: - query = "/rest/accounts/{0}/transactions?{1}".format(account_id, params) + path = "/rest/accounts/{0}/transactions?{1}".format(account_id, urllib.urlencode(options)) else: - query = "/rest/transactions?{0}".format(params) + path = "/rest/transactions?{0}".format( urllib.urlencode(options)) - return self._query_api_object(Transaction, query, collection_name="transactions") + return self._query_api_object(Transaction, path, collection_name="transactions") - def get_transaction(self, account_or_account_id, transaction_id): + def get_transaction(self, account_or_account_id, transaction_id, cents): """Retrieve a specific transaction. Args: account_or_account_id: account to be queried or its ID transaction_id: ID of the transaction to be retrieved + cents (bool): If true amounts will be shown in cents, Optional, default: False Returns: a Transaction object representing the transaction to be retrieved """ - if isinstance(account_or_account_id, Account): - account_or_account_id = account_or_account_id.account_id + options = { "cents": cents } if cents else {} + + account_id = getAccountId(account_or_account_id) + if account_id is not None: + path = "/rest/accounts/{0}/transactions/{1}?{2}".format(account_or_account_id, transaction_id, urllib.urlencode(options)) + else: + path = "/rest/transactions/{0}?{1}".format(transaction_id, urllib.urlencode(options)) - query = "/rest/accounts/{0}/transactions/{1}".format(account_or_account_id, transaction_id) - return self._query_api_object(Transaction, query) + return self._query_api_object(Transaction, path) @property def securities(self): """An array of `Security` objects, one for each transaction of the user.""" return self._query_api_object(Security, "/rest/securities", collection_name="securities") - def get_securities(self, account_id=None, since=None, count=1000, offset=0, accounts=None): + def get_securities(self, account_or_account_id=None, since=None, count=1000, offset=0, accounts=None): """Get an array of `Security` objects, one for each security of the user. Args: - account_id: ID of the account for which to list the securities + account_id: Account for which to list the securities or its ID since: this parameter can either be a transaction ID or a date count: limit the number of returned transactions offset: which offset into the result set should be used to determine the first @@ -1034,7 +1235,7 @@ def get_securities(self, account_id=None, since=None, count=1000, offset=0, acco params['since'] = since params = urllib.urlencode(params) - + account_id = getAccountId(account_or_account_id) if account_id: query = "/rest/accounts/{0}/securities?{1}".format(account_id, params) else: @@ -1052,10 +1253,8 @@ def get_security(self, account_or_account_id, security_id): Returns: a Security object representing the transaction to be retrieved """ - if isinstance(account_or_account_id, Account): - account_or_account_id = account_or_account_id.account_id - query = "/rest/accounts/{0}/securities/{1}".format(account_or_account_id, security_id) + query = "/rest/accounts/{0}/securities/{1}".format(getAccountId(account_or_account_id), security_id) return self._query_api_object(Security, query) def modify_security(self, account_or_account_id, security_or_security_id, visited=None): @@ -1210,7 +1409,7 @@ def modify_user(self, user): def remove_user(self): """Delete figo Account.""" - self._request_with_exception("/rest/user", method="DELETE") + return self._request_with_exception("/rest/user", data=None, method="DELETE") def get_sync_url(self, state, redirect_uri): """URL to trigger a synchronization. diff --git a/figo/models.py b/figo/models.py index 91a3281..e7b8969 100644 --- a/figo/models.py +++ b/figo/models.py @@ -294,6 +294,61 @@ def __unicode__(self): return u"Payment: %s (%s at %s)" % (self.name, self.account_number, self.bank_name) +class StandingOrder(ModelBase): + """Object representing one standing order on a certain bank account of the user. + + Attributes: + standing_order_id: internal figo stanging order id + account_id: internal figo account id + iban: iban of creditor or debtor + amount: order amount + currency: three character currency code + cents: + name: name of originator or recipient + purpose: purpose text + execution_day: number of days of execution of the standing order + first_execution_date: starting day of execution + last_execution_date: finishing day of the execution + interval: + created_at: internal creation timestamp + modified_at: internal creation timestamp + """ + + __dump_attributes__ = [] + + standing_order_id = None + account_id = None + iban = None + amount = None + currency = None + cents = None + name = None + purpose = None + execution_day = None + first_execution_date = None + last_execution_date = None + interval = None + created_at = None + modified_at = None + + def __init__(self, session, **kwargs): + super(StandingOrder, self).__init__(session, **kwargs) + + if self.created_at: + self.created_at = dateutil.parser.parse(self.created_at) + + if self.modified_at: + self.modified_at = dateutil.parser.parse(self.modified_at) + + if self.first_execution_date: + self.first_execution_date = dateutil.parser.parse(self.first_execution_date) + + if self.last_execution_date: + self.last_execution_date = dateutil.parser.parse(self.last_execution_date) + + def __unicode__(self): + return u"Standing Order: %s " % (self.id) + class Transaction(ModelBase): """Object representing one bank transaction on a certain bank account of the user. @@ -482,6 +537,44 @@ class SynchronizationStatus(ModelBase): def __unicode__(self): return u"Synchronization Status: %s (%s)" % (self.code, self.message) +class Sync(ModelBase): + """Object representing a syncronisation for account creation. + + Attributes: + id: internal figo syncronisation id + status: Current processing state of the item. + challenge: AuthMethodSelectChallenge (object) or EmbeddedChallenge (object) or RedirectChallenge (object) or DecoupledChallenge (object) (Challenge). + error: Error detailing why the background operation failed. + created_at: Time at which the sync was created + started_at: Time at which the sync started + ended_at: Time at which the sync ended + """ + __dump_attributes__ = [] + + id = None + status = None + challenge = None + error = None + created_at = None + started_at = None + ended_at = None + + def __init__(self, session, **kwargs): + super(Sync, self).__init__(session, **kwargs) + if self.created_at: + self.created_at = dateutil.parser.parse(self.created_at) + + if self.started_at: + self.started_at = dateutil.parser.parse(self.started_at) + + if self.ended_at: + self.ended_at = dateutil.parser.parse(self.ended_at) + + if self.challenge: + self.challenge = Challenge.from_dict(self.session, self.challenge) + + def __unicode__(self): + return u"Sync: %s" % (self.id) class User(ModelBase): """Object representing an user. @@ -569,8 +662,8 @@ class Service(ModelBase): def __init__(self, session, **kwargs): super(Service, self).__init__(session, **kwargs) if self.language: - self.available_languages = [l for l in self.language['available_languages']] - self.language = self.language['current_language'] + self.available_languages = [l for l in self.language['available']] + self.language = self.language['current'] def __unicode__(self, *args, **kwargs): return u"Service: %s" % (self.bank_code) @@ -685,10 +778,12 @@ class Challenge(ModelBase): """ __dump_attributes__ = ["title", "label", "format"] + id = None title = None label = None format = None data = None + type = None def __unicode__(self, *args, **kwargs): return u"Challenge: %s" % (self.title) diff --git a/requirements.txt b/requirements.txt index db68a22..d3fc3d7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,4 @@ python-dateutil requests requests_toolbelt setuptools_scm +python-dotenv diff --git a/tests/conftest.py b/tests/conftest.py index 4d2a655..842314e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,15 +1,22 @@ import pytest import uuid import time +import os from logging import basicConfig -from figo.credentials import CREDENTIALS +# from figo.credentials import CREDENTIALS from figo import FigoConnection from figo import FigoSession +from dotenv import load_dotenv +load_dotenv() + basicConfig(level='DEBUG') +API_ENDPOINT = os.getenv("API_ENDPOINT") +CLIENT_ID = os.getenv("CLIENT_ID") +CLIENT_SECRET = os.getenv("CLIENT_SECRET") PASSWORD = 'some_words' @@ -20,66 +27,22 @@ def new_user_id(): @pytest.fixture(scope='module') def figo_connection(): - return FigoConnection(CREDENTIALS['client_id'], - CREDENTIALS['client_secret'], + return FigoConnection(CLIENT_ID, + CLIENT_SECRET, "https://127.0.0.1/", - api_endpoint=CREDENTIALS['api_endpoint']) + api_endpoint=API_ENDPOINT) @pytest.fixture(scope='module') def figo_session(figo_connection, new_user_id): - figo_connection.add_user("Test", new_user_id, PASSWORD) - response = figo_connection.credential_login(new_user_id, PASSWORD) - - scope = response['scope'] - - required_scopes = [ - 'accounts=rw', - 'transactions=rw', - 'user=rw', - 'create_user', - ] - - if any(s not in scope for s in required_scopes): - pytest.skip("The client ID needs write access to the servers.") - - session = FigoSession(response['access_token']) - - task_token = session.add_account("de", ("figo", "figo"), "90090042") - state = session.get_task_state(task_token) - - while not (state.is_ended or state.is_erroneous): - state = session.get_task_state(task_token) - time.sleep(2) - assert not state.is_erroneous - - yield session - - session.remove_user() - - -@pytest.fixture(scope='module') -def account_ids(figo_session): - accs = figo_session.accounts - - yield [a.account_id for a in accs] - - -@pytest.fixture(scope='module') -def giro_account(figo_session): - # returns the first account from the demo bank that is of type "Girokonto" - # and asserts there is at least one - accs = figo_session.accounts - giro_accs = [a for a in accs if a.type == "Giro account"] - assert len(giro_accs) >= 1 - - yield giro_accs[0] - + figo_connection.add_user("Test", new_user_id, PASSWORD) + response = figo_connection.credential_login(new_user_id, PASSWORD) + return FigoSession(response['access_token']) @pytest.fixture(scope='module') def access_token(figo_connection, new_user_id): figo_connection.add_user("Test", new_user_id, PASSWORD) - response = figo_connection.credential_login(new_user_id, PASSWORD) + response = figo_connection.credential_login(new_user_id, PASSWORD, scope="user=rw accounts=rw transactions=rw") access_token = response['access_token'] yield access_token diff --git a/tests/test_catalog_and_language.py b/tests/test_catalog_and_language.py index 94f427e..a85d4c6 100644 --- a/tests/test_catalog_and_language.py +++ b/tests/test_catalog_and_language.py @@ -1,39 +1,56 @@ import pytest +import os from figo import FigoException +from figo import FigoConnection from figo import FigoSession from figo.models import Service from figo.models import LoginSettings +from figo.models import BankContact + +from dotenv import load_dotenv +load_dotenv() + +API_ENDPOINT = os.getenv("API_ENDPOINT") +CLIENT_ID = os.getenv("CLIENT_ID") +CLIENT_SECRET = os.getenv("CLIENT_SECRET") -CREDENTIALS = ["figo", "figo"] -BANK_CODE = "90090042" CLIENT_ERROR = 1000 +connection = FigoConnection(CLIENT_ID, CLIENT_SECRET, "https://127.0.0.1/", api_endpoint=API_ENDPOINT) +@pytest.mark.parametrize('language', ['de']) +@pytest.mark.parametrize('country', ['DE', 'AT']) +def test_get_catalog_en_client_auth(language, country): + catalog = connection.get_catalog(None, country) + for bank in catalog['banks']: + assert isinstance(bank, BankContact) + assert bank.country == country + for service in catalog['services']: + assert isinstance(service, Service) + +def test_get_catalog_client_auth_query(): + q = 'PayPal' + catalog = connection.get_catalog(q, None) + for bank in catalog['banks']: + assert bank.name == q + for service in catalog['services']: + assert service.name == q -@pytest.mark.parametrize('language', ['de', 'en']) -def test_get_catalog_en(access_token, language): +@pytest.mark.parametrize('language', ['de']) +@pytest.mark.parametrize('country', ['DE', 'FR']) +def test_get_catalog_en(access_token, language, country): figo_session = FigoSession(access_token) figo_session.language = language - catalog = figo_session.get_catalog() + catalog = figo_session.get_catalog(country) for bank in catalog['banks']: - assert bank.language == language - + assert bank.country == country def test_get_catalog_invalid_language(access_token): figo_session = FigoSession(access_token) - figo_session.language = 'xy' with pytest.raises(FigoException) as e: - figo_session.get_catalog() + figo_session.get_catalog("XY") assert e.value.code == CLIENT_ERROR - -def test_get_supported_payment_services(access_token): - figo_session = FigoSession(access_token) - services = figo_session.get_supported_payment_services("de") - assert len(services) > 10 # this a changing value, this tests that at least some are returned - assert isinstance(services[0], Service) - - # XXX(Valentin): Catalog needs `accounts=rw`, so it doesn't work with the demo session. # Sounds silly at first, but actually there is no point to view the catalog if # you can't add accounts. @@ -41,21 +58,3 @@ def test_get_catalog(access_token): figo_session = FigoSession(access_token) catalog = figo_session.get_catalog() assert len(catalog) == 2 - - -def test_get_login_settings(access_token): - figo_session = FigoSession(access_token) - login_settings = figo_session.get_login_settings("de", BANK_CODE) - assert isinstance(login_settings, LoginSettings) - assert login_settings.advice - assert login_settings.credentials - - -def test_set_unset_language(access_token): - figo_session = FigoSession(access_token) - assert figo_session.language is None - figo_session.language = 'de' - assert figo_session.language == 'de' - figo_session.language = '' - assert figo_session.language is None - figo_session.language = 'de' diff --git a/tests/test_integrations.py b/tests/test_integrations.py new file mode 100644 index 0000000..d7c1b67 --- /dev/null +++ b/tests/test_integrations.py @@ -0,0 +1,144 @@ +import pytest +import time +import os + +from figo.models import User +from figo.models import TaskState +from figo.models import TaskToken +from figo.models import Account +from figo.models import Payment +from figo.models import Sync +from figo.models import Challenge + +from figo import FigoConnection +from figo import FigoSession +from figo import FigoException + +from dotenv import load_dotenv +load_dotenv() + +API_ENDPOINT = os.getenv("API_ENDPOINT") +CLIENT_ID = os.getenv("CLIENT_ID") +CLIENT_SECRET = os.getenv("CLIENT_SECRET") + +connection = FigoConnection(CLIENT_ID, CLIENT_SECRET, "https://127.0.0.1/", api_endpoint=API_ENDPOINT) +CREDENTIALS = { 'account_number' : "foobarbaz", 'pin' : "12345" } +CONSENT = { "recurring": True, "period": 90, "scopes": ["ACCOUNTS", "BALANCES", "TRANSACTIONS"], "accounts": [{ "id": "DE67900900424711951500", "currency": "EUR" }] } +ACCESS_METHOD_ID = "ae441170-b726-460c-af3c-b76756de00e0" +data = {} + +def pytest_namespace(): + return {'session': '', 'token': '', 'access_id': '', 'sync_id': '', 'challenge_id': '', 'account_id': '', 'payments_token': ''} + +def test_add_user(): + response = connection.add_user("John Doe", "john.doe@example.com", "password") + assert response == {} + +def test_get_version(): + response = connection.get_version() + assert response == {'environment': 'staging', 'version': '19.8.0.0rc46'} + +def test_create_token_and_session(): + token = connection.credential_login("john.doe@example.com", "password") + pytest.token = token["access_token"] + + pytest.session = FigoSession(pytest.token) + assert pytest.session.user.full_name == "John Doe" + +def test_create_token_for_payments(): + token = connection.credential_login("john.doe@example.com", "password", scope="payments=rw") + pytest.payments_token = token["access_token"] + assert token["scope"] == "payments=rw" + +def test_add_access(): + response = pytest.session.add_access(ACCESS_METHOD_ID, CREDENTIALS, CONSENT) + pytest.access_id = response["id"] + assert response.has_key("id") == True + +def test_add_access_with_wrong_access_id(access_token): + figo_session = FigoSession(access_token) + access_method_id = "pipo" + response = figo_session.add_access(access_method_id,CREDENTIALS,CONSENT) + assert response.has_key("error") == True + +def test_get_accesses(): + accesses = pytest.session.get_accesses() + assert len(accesses) > 0 + +def test_get_access(): + accesses = pytest.session.get_access(pytest.access_id) + assert len(accesses) > 0 + +def test_add_sync(): + response = pytest.session.add_sync(pytest.access_id, None, None, None, None, None) + pytest.sync_id = response.id + assert isinstance(response, Sync) + assert response.status == 'QUEUED' + +def test_get_synchronization_status(): + time.sleep(10) + response = pytest.session.get_synchronization_status(pytest.access_id, pytest.sync_id) + pytest.challenge_id = response.challenge.id + assert isinstance(response, Sync) + assert response.status == "AWAIT_AUTH" + +def test_solve_synchronization_challenge(access_token): + payload = { "value": "111111" } + response = pytest.session.solve_synchronization_challenge(pytest.access_id, pytest.sync_id, pytest.challenge_id, payload) + assert response == {} + +def test_get_sync_after_challenge(): + time.sleep(10) + response = pytest.session.get_synchronization_status(pytest.access_id, pytest.sync_id) + assert isinstance(response, Sync) + assert response.status == "COMPLETED" or response.status == "RUNNING" + +def test_get_synchronization_challenges(): + response = pytest.session.get_synchronization_challenges(pytest.access_id, pytest.sync_id) + assert len(response) > 0 + +def test_get_synchronization_challenge(): + response = pytest.session.get_synchronization_challenge(pytest.access_id, pytest.sync_id, pytest.challenge_id) + assert isinstance(response, Challenge) + +def test_get_accounts(): + response = pytest.session.get_accounts() + pytest.account_id = response[0].account_id + assert isinstance(response[0], Account) + assert isinstance(response[0].account_id, unicode) + +def test_get_account(): + response = pytest.session.get_account(pytest.account_id) + assert isinstance(response, Account) + +def test_get_account_balance(): + response = pytest.session.get_account_balance(pytest.account_id) + assert response.balance == 0 + +def test_get_securities(access_token): + response = pytest.session.get_securities() + assert response != None + +def test_get_payments(): + session = FigoSession(pytest.payments_token) + response = session.get_payments(pytest.account_id, None, None, None, None) + assert response == [] + +def test_get_standing_orders(): + response = pytest.session.get_standing_orders() + assert response == [] + +#todo: check response API +def test_remove_pin(): + response = pytest.session.remove_pin(pytest.access_id) + assert response != None + +def test_delete_account(): + response = pytest.session.remove_account(pytest.account_id) + assert response == None + +def test_remove_user(): + response = pytest.session.remove_user() + assert response == {} + + diff --git a/tests/test_models.py b/tests/test_models.py index 0b9f2bd..557fac8 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -16,16 +16,15 @@ from figo.models import ProcessStep from figo.models import Security from figo.models import Service +from figo.models import StandingOrder from figo.models import SynchronizationStatus from figo.models import TaskState from figo.models import TaskToken from figo.models import Transaction from figo.models import User -from tests.test_writing_methods import CLIENT_ERROR - HTTP_NOT_ACCEPTABLE = 406 - +CLIENT_ERROR = 1000 def test_create_account_from_dict(figo_session): data = { @@ -132,6 +131,25 @@ def test_create_transaction_from_dict(figo_session): transaction = Transaction.from_dict(figo_session, data) assert isinstance(transaction, Transaction) +def test_create_standing_order_from_dict(figo_session): + data = { + "account_id": "A12345.6", + "standing_order_id": "SO12345.6", + "iban": "DE99012345678910020030", + "amount": 125.5, + "currency": "EUR", + "cents": False, + "name": "John Doe", + "purpose": "So long and thanks for all the fish", + "execution_day": 1, + "first_execution_date": "2018-08-30T00:00:00.000Z", + "last_execution_date": "2018-08-30T00:00:00.000Z", + "interval": "monthly", + "created_at": "2018-08-30T00:00:00.000Z", + "modified_at": "2018-08-31T00:00:00.000Z" + } + standing_order = StandingOrder.from_dict(figo_session, data) + assert isinstance(standing_order, StandingOrder) def test_create_transaction_with_categories(figo_session): data = { @@ -228,11 +246,11 @@ def test_create_service_from_dict(figo_session): "icon": "https://api.figo.me/assets/images/accounts/demokonto.png", "name": "Demokonto", "language": { - "available_languages": [ + "available": [ "de", "en", ], - "current_language": "de", + "current": "de", }, } service = Service.from_dict(figo_session, data) diff --git a/tests/test_notifications.py b/tests/test_notifications.py new file mode 100644 index 0000000..6c27849 --- /dev/null +++ b/tests/test_notifications.py @@ -0,0 +1,56 @@ +import pytest +import os + +from figo import FigoConnection +from figo import FigoSession +from figo.models import Notification + +API_ENDPOINT = os.getenv("API_ENDPOINT") +CLIENT_ID = os.getenv("CLIENT_ID") +CLIENT_SECRET = os.getenv("CLIENT_SECRET") +connection = FigoConnection(CLIENT_ID, CLIENT_SECRET, "https://127.0.0.1/", api_endpoint=API_ENDPOINT) + +def pytest_namespace(): + return {'session': '', 'token': '', 'access_id': '', 'sync_id': '', 'challenge_id': ''} + +#To call only once +def test_add_user(): + response = connection.add_user("John Doe", "john.doe@example.com", "password") + assert response == {} + +def test_create_token_and_session(): + token = connection.credential_login("john.doe@example.com", "password") + pytest.token = token["access_token"] + assert pytest.token + pytest.session = FigoSession(token["access_token"]) + assert pytest.session.user.full_name == "John Doe" + +def test_add_notification(access_token): + notification = Notification(pytest.session) + notification.notify_uri = "https://api.figo.me/callback" + notification.observe_key = "/rest/transactions" + notification.state = "4HgwtQP0jsjdz79h" + response = pytest.session.add_notification(notification) + pytest.notification = response + assert response.notification_id != None + +def test_notifications(access_token): + response = pytest.session.notifications + assert response != None + +def test_modify_notification(access_token): + pytest.notification.state="ZZZ" + response = pytest.session.modify_notification(pytest.notification) + assert response.state != None + +def test_get_notification(access_token): + response = pytest.session.get_notification(pytest.notification.notification_id) + assert response.state != None + +def test_remove_notification(access_token): + response = pytest.session.remove_notification(pytest.notification.notification_id) + assert response == None + +def test_remove_user(): + response = pytest.session.remove_user() + assert response == {} diff --git a/tests/test_session.py b/tests/test_session.py deleted file mode 100644 index 733b62e..0000000 --- a/tests/test_session.py +++ /dev/null @@ -1,177 +0,0 @@ -from __future__ import unicode_literals - -import platform - -import pytest - -from figo import FigoException -from figo.models import Notification -from figo.models import Payment -from figo.models import TaskToken - - -def test_get_account(figo_session, account_ids): - account_id = account_ids[0] - account = figo_session.get_account(account_id) - assert account.account_id == account_id - - -def test_get_account_tan_schemes(figo_session, giro_account): - account = figo_session.get_account(giro_account.account_id) - assert len(account.supported_tan_schemes) > 0 - - -def test_get_account_balance(figo_session, giro_account): - # account sub-resources - balance = figo_session.get_account_balance(figo_session.get_account(giro_account.account_id)) - assert balance.balance - assert balance.balance_date - - -def test_get_account_transactions(figo_session, giro_account): - transactions = figo_session.get_account(giro_account.account_id).transactions - assert len(transactions) > 0 - - -def test_get_account_payments(figo_session, giro_account): - payments = figo_session.get_account(giro_account.account_id).payments - assert len(payments) >= 0 - - -def test_get_global_transactions(figo_session): - transactions = figo_session.transactions - assert len(transactions) > 0 - - -def test_get_global_payments(figo_session): - payments = figo_session.payments - assert len(payments) >= 0 - - -def test_get_notifications(figo_session): - notifications = figo_session.notifications - assert len(notifications) >= 0 - - -def test_get_missing_account(figo_session): - with pytest.raises(FigoException): - figo_session.get_account("A1.22") - - -def test_error_handling(figo_session): - with pytest.raises(FigoException): - figo_session.get_sync_url('qwe', 'qew') - - -def test_sync_uri(figo_session): - figo_session.get_sync_url('some_state', 'http://example.com') - - -def test_get_mail_from_user(figo_session): - assert figo_session.user.email.endswith("testuser@example.com") - - -@pytest.mark.skip(reason="race condition on travis") -def test_create_update_delete_notification(figo_session): - """ - This test sometimes fails, when run for different versions in parallel, e.g. on travis - It happens because the notification id will always be the same for the demo client. - This will be solved with running tests against an enhanced sandbox. - """ - state_version = "V{0}".format(platform.python_version()) - added_notification = figo_session.add_notification( - Notification.from_dict(figo_session, dict(observe_key="/rest/transactions", - notify_uri="http://figo.me/test", - state=state_version))) - - assert added_notification.observe_key == "/rest/transactions" - assert added_notification.notify_uri == "http://figo.me/test" - assert added_notification.state == state_version - - print("\n##############") - print("id: {0}, {1}".format(added_notification.notification_id, added_notification.state)) - - added_notification.state = state_version + "_modified" - modified_notification = figo_session.modify_notification(added_notification) - assert modified_notification.observe_key == "/rest/transactions" - assert modified_notification.notify_uri == "http://figo.me/test" - assert modified_notification.state == state_version + "_modified" - - print("id: {0}, {1}".format(modified_notification.notification_id, modified_notification.state)) - - figo_session.remove_notification(modified_notification.notification_id) - with pytest.raises(FigoException): - deleted_notification = figo_session.get_notification(modified_notification.notification_id) - print("id: {0}, {1}".format( - deleted_notification.notification_id, deleted_notification.state)) - print("#"*10) - - -def test_create_update_delete_payment(figo_session, giro_account): - added_payment = figo_session.add_payment( - Payment.from_dict(figo_session, dict(account_id=giro_account.account_id, - type="Transfer", - account_number="4711951501", - bank_code="90090042", - name="figo", - purpose="Thanks for all the fish.", - amount=0.89))) - - assert added_payment.account_id, giro_account.account_id - assert added_payment.bank_name == "Demobank" - assert added_payment.amount == 0.89 - - added_payment.amount = 2.39 - modified_payment = figo_session.modify_payment(added_payment) - assert modified_payment.payment_id == added_payment.payment_id - assert modified_payment.account_id == giro_account.account_id - assert modified_payment.bank_name == "Demobank" - assert modified_payment.amount == 2.39 - - figo_session.remove_payment(modified_payment) - with pytest.raises(FigoException): - figo_session.get_payment(modified_payment.account_id, modified_payment.payment_id) - - -def test_delete_transaction(figo_session, giro_account): - transaction = giro_account.transactions[0] - figo_session.delete_transaction(giro_account.account_id, transaction.transaction_id) - - -def test_get_payment_proposals(figo_session): - proposals = figo_session.get_payment_proposals() - assert len(proposals) >= 1 - - -def test_start_task(figo_session): - # Valid task token needed - task_token = TaskToken(figo_session) - task_token.task_token = "invalidTaskToken" - with pytest.raises(FigoException): - figo_session.start_task(task_token) - - -def test_poll_task_state(figo_session): - # Valid task token needed - task_token = TaskToken(figo_session) - task_token.task_token = "invalidTaskToken" - with pytest.raises(FigoException): - figo_session.get_task_state(task_token) - - -def test_cancel_task(figo_session): - # Valid task token needed - task_token = TaskToken(figo_session) - task_token.task_token = "invalidTaskToken" - with pytest.raises(FigoException): - figo_session.cancel_task(task_token) - - -def test_sync_account(figo_session): - assert figo_session.sync_account(state="qweqwe") - - -def test_get_bank(figo_session, giro_account): - - bank = figo_session.get_bank(giro_account.bank_id) - assert bank.bank_id diff --git a/tests/test_writing_methods.py b/tests/test_writing_methods.py deleted file mode 100644 index b2dc6d3..0000000 --- a/tests/test_writing_methods.py +++ /dev/null @@ -1,142 +0,0 @@ -# coding:utf-8 - -import pytest -import time - -from mock import patch - -from figo import FigoException -from figo import FigoPinException - -from figo.models import TaskState -from figo.models import TaskToken - - -CREDENTIALS = ["figo", "figo"] -BANK_CODE = "90090042" -CLIENT_ERROR = 1000 - - -def test_add_account(figo_session): - token = figo_session.add_account("de", CREDENTIALS, BANK_CODE) - assert isinstance(token, TaskToken) - task_state = figo_session.get_task_state(token) - time.sleep(5) - assert isinstance(task_state, TaskState) - assert len(figo_session.accounts) >= 1 - - -def test_add_account_and_sync_wrong_pin(figo_session): - wrong_credentials = [CREDENTIALS[0], "123456"] - try: - with pytest.raises(FigoException): - figo_session.add_account_and_sync("de", wrong_credentials, BANK_CODE) - except FigoException as figo_exception: - # BBB(Valentin): prevent demo account from complaining - it returns no code on error - if "Please use demo account credentials" not in figo_exception.error_description: - raise - - -def test_add_account_and_sync_wrong_pin_postbank(figo_session): - """ - Check that `FigoPinException` is raised correctly on given task state, which occurs - when attempting to add an account to Postbank with syntactically correct (9-digit login), but - invalid credentials. Note that syntactically incorrect credentials return code `20000` and a - different message. - """ - - mock_task_state = { - "is_ended": True, - "account_id": u"A2248267.0", - "is_waiting_for_pin": False, - "is_erroneous": True, - "message": u"Die Anmeldung zum Online-Zugang Ihrer Bank ist fehlgeschlagen. " - u"Bitte überprüfen Sie Ihre Benutzerkennung.", - "error": { - "code": 10000, - "group": u"user", - "name": u"Login credentials are invalid", - "message": u"9050 Die Nachricht enthält Fehler.; 9800 Dialog abgebrochen; " - u"9010 Initialisierung fehlgeschlagen, Auftrag nicht bearbeitet.; " - u"3920 Zugelassene Zwei-Schritt-Verfahren für den Benutzer.; " - u"9010 PIN/TAN Prüfung fehlgeschlagen; " - u"9931 Anmeldename oder PIN ist falsch.", - "data": {}, - "description": u"Die Anmeldung zum Online-Zugang Ihrer Bank ist fehlgeschlagen. " - u"Bitte überprüfen Sie Ihre Benutzerkennung." - }, - "challenge": {}, - "is_waiting_for_response": False - } - - with patch.object(figo_session, 'get_task_state') as mock_state: - with patch.object(figo_session, 'add_account') as mock_account: - - mock_state.return_value = TaskState.from_dict(figo_session, mock_task_state) - mock_account.return_value = None - - with pytest.raises(FigoPinException) as e: - figo_session.add_account_and_sync("de", None, None) - assert e.value.code == 10000 - - -@pytest.mark.skip(reason="test is flaky as hell and should be rewritten completely") -def test_add_account_and_sync_wrong_and_correct_pin(figo_session): - wrong_credentials = [CREDENTIALS[0], "123456"] - figo_session.sync_poll_retry = 100 - try: - task_state = figo_session.add_account_and_sync("de", wrong_credentials, BANK_CODE) - except FigoPinException as pin_exception: - task_state = figo_session.add_account_and_sync_with_new_pin(pin_exception, CREDENTIALS[1]) - assert isinstance(task_state, TaskState) - assert len(figo_session.accounts) == 3 - except FigoException as figo_exception: - # XXXValentin): prevent demo account from complaining - if figo_exception.code != 90000: - raise - - -@pytest.mark.skip(reason="test expects state of account, that are not prepared at the moment") -def test_modify_transaction(figo_session): - account = figo_session.accounts[0] - transaction = account.transactions[0] - response = figo_session.modify_transaction( - account.account_id, - transaction.transaction_id, - visited=False) - - assert not response.visited - response = figo_session.modify_transaction( - account.account_id, - transaction.transaction_id, - visited=True) - - assert response.visited - - -@pytest.mark.skip(reason="test expects state of account, that are not prepared at the moment") -def test_modify_account_transactions(figo_session): - account = figo_session.accounts[0] - figo_session.modify_account_transactions(account.account_id, False) - - assert not any([transaction.visited for transaction in account.transactions]) - figo_session.modify_account_transactions(account.account_id, True) - assert all([transaction.visited for transaction in account.transactions]) - - -@pytest.mark.skip(reason="test expects state of account, that are not prepared at the moment") -def test_modify_user_transactions(figo_session): - figo_session.modify_user_transactions(False) - assert not any([transaction.visited for transaction in figo_session.transactions]) - - figo_session.modify_user_transactions(True) - assert all([transaction.visited for transaction in figo_session.transactions]) - - -@pytest.mark.skip(reason="test expects state of account, that are not prepared at the moment") -def test_delete_transaction(figo_session): - account = figo_session.accounts[0] - transaction = account.transactions[0] - transaction_count = len(account.transactions) - figo_session.delete_transaction(account.account_id, transaction.transaction_id) - assert transaction_count - 1 == account.transactions