diff options
author | MS <ms@taler.net> | 2021-01-21 16:44:15 +0100 |
---|---|---|
committer | MS <ms@taler.net> | 2021-01-21 16:44:15 +0100 |
commit | 80478f3885f2b3c56ed7d7b0127784549d75c7d2 (patch) | |
tree | 8cb7fecbcb5643328a41ceac6b7f348e2b487d1c | |
parent | df7d8c253147b83cfcfbc962c1a244f8fe8060bf (diff) | |
download | libeufin-80478f3885f2b3c56ed7d7b0127784549d75c7d2.tar.gz libeufin-80478f3885f2b3c56ed7d7b0127784549d75c7d2.tar.bz2 libeufin-80478f3885f2b3c56ed7d7b0127784549d75c7d2.zip |
remove unwanted
-rw-r--r-- | build.gradle | 10 | ||||
-rwxr-xr-x | cli/bin/libeufin-cli/libeufin-cli.template | 1048 |
2 files changed, 10 insertions, 1048 deletions
diff --git a/build.gradle b/build.gradle index a4b74272..61ba290a 100644 --- a/build.gradle +++ b/build.gradle @@ -1,3 +1,5 @@ +import org.apache.tools.ant.filters.ReplaceTokens + plugins { id 'org.jetbrains.kotlin.jvm' version '1.4.30-RC' id 'idea' @@ -28,12 +30,20 @@ task versionFile() { new File("${projectDir}/util/src/main/resources", "version.txt").text = getRootProject().version } +task replaceVersionCli(type: Copy) { + from file("cli/bin/libeufin-cli.template") + into file("cli/bin/libeufin-cli") + filter(ReplaceTokens, tokens: [version: getRootProject().version]) +} + classes { dependsOn versionFile + dependsOn replaceVersionCli } task dist(type: Zip) { dependsOn versionFile + dependsOn replaceVersionCli evaluationDependsOn("nexus") evaluationDependsOn("sandbox") def topDir = "${getRootProject().name}-${getRootProject().version}" diff --git a/cli/bin/libeufin-cli/libeufin-cli.template b/cli/bin/libeufin-cli/libeufin-cli.template deleted file mode 100755 index c7aabe54..00000000 --- a/cli/bin/libeufin-cli/libeufin-cli.template +++ /dev/null @@ -1,1048 +0,0 @@ -#!/usr/bin/env python3 - -import os -import sys -import click -import json -import hashlib -import errno -from datetime import datetime -from requests import post, get, auth, delete -from urllib.parse import urljoin -from getpass import getpass - - -def tell_user(resp, withsuccess=False): - if resp.status_code == 200 and not withsuccess: - return - print(resp.content.decode("utf-8")) - - -# FIXME: deprecate this in favor of NexusContext -def fetch_env(): - if "--help" in sys.argv: - return [] - try: - nexus_base_url = os.environ["LIBEUFIN_NEXUS_URL"] - nexus_username = os.environ["LIBEUFIN_NEXUS_USERNAME"] - nexus_password = os.environ["LIBEUFIN_NEXUS_PASSWORD"] - except KeyError: - print( - "Please ensure that NEXUS_BASE_URL," - " NEXUS_USERNAME, NEXUS_PASSWORD exist" - " in the environment" - ) - sys.exit(1) - return nexus_base_url, nexus_username, nexus_password - - -# FIXME: deprecate this in favor of NexusContext -class NexusAccess: - def __init__(self, nexus_base_url=None, username=None, password=None): - self.nexus_base_url = nexus_base_url - self.username = username - self.password = password - - -@click.group(help="General utility to invoke HTTP REST services offered by Nexus.") -@click.version_option(version="0.0.1-dev.1") -def cli(): - pass - - -@cli.group() -@click.pass_context -def facades(ctx): - ctx.obj = NexusAccess(*fetch_env()) - - -@cli.group() -@click.pass_context -def connections(ctx): - ctx.obj = NexusAccess(*fetch_env()) - - -@cli.group() -@click.pass_context -def users(ctx): - ctx.obj = NexusContext() - -@cli.group() -@click.pass_context -def permissions(ctx): - ctx.obj = NexusContext() - -@users.command("list", help="List users") -@click.pass_obj -def list_users(obj): - url = urljoin(obj.nexus_base_url, f"/users") - try: - resp = get(url, auth=auth.HTTPBasicAuth(obj.nexus_username, obj.nexus_password)) - except Exception as e: - print(e) - print("Could not reach nexus at " + url) - exit(1) - - print(resp.content.decode("utf-8")) - - -@users.command("create", help="Create a new user") -@click.argument("username") -@click.option( - "--password", - help="Provide password instead of prompting interactively.", - prompt=True, - hide_input=True, - confirmation_prompt=True, -) -@click.pass_obj -def create_user(obj, username, password): - url = urljoin(obj.nexus_base_url, f"/users") - try: - body = dict( - username=username, - password=password, - ) - resp = post( - url, - json=body, - auth=auth.HTTPBasicAuth(obj.nexus_username, obj.nexus_password), - ) - except Exception as e: - print(e) - print("Could not reach nexus at " + url) - exit(1) - - print(resp.content.decode("utf-8")) - - -@permissions.command("list", help="Show permissions") -@click.pass_obj -def list_permission(obj): - url = urljoin(obj.nexus_base_url, f"/permissions") - try: - resp = get(url, auth=auth.HTTPBasicAuth(obj.nexus_username, obj.nexus_password)) - except Exception as e: - print(e) - print("Could not reach nexus at " + url) - exit(1) - - print(resp.content.decode("utf-8")) - -@permissions.command("grant", help="Grant permission to a subject") -@click.pass_obj -@click.argument("subject-type") -@click.argument("subject-id") -@click.argument("resource-type") -@click.argument("resource-id") -@click.argument("permission-name") -def grant_permission(obj, subject_type, subject_id, resource_type, resource_id, permission_name): - url = urljoin(obj.nexus_base_url, f"/permissions") - try: - permission = dict( - subjectType=subject_type, - subjectId=subject_id, - resourceType=resource_type, - resourceId=resource_id, - permissionName=permission_name, - ) - body = dict( - permission=permission, - action="grant", - ) - resp = post(url, json=body, auth=auth.HTTPBasicAuth(obj.nexus_username, obj.nexus_password)) - except Exception as e: - print(e) - print("Could not reach nexus at " + url) - exit(1) - - print(resp.content.decode("utf-8")) - -@permissions.command("revoke", help="Revoke permission from a subject") -@click.pass_obj -@click.argument("subject-type") -@click.argument("subject-id") -@click.argument("resource-type") -@click.argument("resource-id") -@click.argument("permission-name") -def grant_permission(obj, subject_type, subject_id, resource_type, resource_id, permission_name): - url = urljoin(obj.nexus_base_url, f"/permissions") - try: - permission = dict( - subjectType=subject_type, - subjectId=subject_id, - resourceType=resource_type, - resourceId=resource_id, - permissionName=permission_name, - ) - body = dict( - permission=permission, - action="revoke", - ) - resp = post(url, json=body, auth=auth.HTTPBasicAuth(obj.nexus_username, obj.nexus_password)) - except Exception as e: - print(e) - print("Could not reach nexus at " + url) - exit(1) - - print(resp.content.decode("utf-8")) - - -@cli.group() -@click.pass_context -def accounts(ctx): - ctx.obj = NexusAccess(*fetch_env()) - - -class SandboxContext: - def __init__(self): - self.sandbox_base_url = None - - def require_sandbox_base_url(self): - if self.sandbox_base_url: - return self.sandbox_base_url - sandbox_base_url = os.environ.get("LIBEUFIN_SANDBOX_URL") - if not sandbox_base_url: - raise click.UsageError( - "sandbox URL must be given as an argument or in LIBEUFIN_SANDBOX_URL" - ) - return sandbox_base_url - - -class NexusContext: - def __init__(self): - self._nexus_base_url = None - self._nexus_password = None - self._nexus_username = None - - @property - def nexus_base_url(self): - if self._nexus_base_url: - return self._nexus_base_url - val = os.environ.get("LIBEUFIN_NEXUS_URL") - if not val: - raise click.UsageError( - "nexus URL must be given as an argument or in LIBEUFIN_NEXUS_URL" - ) - self._nexus_base_url = val - return val - - @property - def nexus_username(self): - if self._nexus_username: - return self._nexus_username - val = os.environ.get("LIBEUFIN_NEXUS_USERNAME") - if not val: - raise click.UsageError( - "nexus username must be given as an argument or in LIBEUFIN_NEXUS_USERNAME" - ) - self._nexus_username = val - return val - - @property - def nexus_password(self): - if self._nexus_password: - return self._nexus_password - val = os.environ.get("LIBEUFIN_NEXUS_PASSWORD") - if not val: - raise click.UsageError( - "nexus password must be given as an argument or in LIBEUFIN_NEXUS_PASSWORD" - ) - self._nexus_password = val - return val - - -@cli.group() -@click.option("--sandbox-url", help="URL for the sandbox", required=False) -@click.pass_context -def sandbox(ctx, sandbox_url): - ctx.obj = SandboxContext() - ctx.obj.sandbox_base_url = sandbox_url - - -@connections.command(help="Get key letter (typically PDF).") -@click.argument("connection-name") -@click.argument("output_file") -@click.pass_obj -def get_key_letter(obj, connection_name, output_file): - url = urljoin(obj.nexus_base_url, f"/bank-connections/{connection_name}/keyletter") - try: - resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password)) - except Exception: - print("Could not reach nexus at " + url) - exit(1) - - if resp.status_code != 200: - print(resp.content.decode("utf-8")) - sys.exit(1) - - output = open(output_file, "wb") - output.write(resp.content) - output.close() - - -@connections.command(help="export backup") -@click.option("--passphrase", help="Passphrase for locking the backup", required=True) -@click.option("--output-file", help="Where to store the backup", required=True) -@click.argument("connection-name") -@click.pass_obj -def export_backup(obj, connection_name, passphrase, output_file): - url = urljoin( - obj.nexus_base_url, "/bank-connections/{}/export-backup".format(connection_name) - ) - try: - resp = post( - url, - json=dict(passphrase=passphrase), - auth=auth.HTTPBasicAuth(obj.username, obj.password), - ) - except Exception: - print("Could not reach nexus at " + url) - exit(1) - - output = open(output_file, "w+") - output.write(resp.content.decode("utf-8")) - output.close() - - print("Backup stored in {}".format(output_file)) - - -@connections.command(help="delete bank connection") -@click.argument("connection-name") -@click.pass_obj -def delete_connection(obj, connection_name): - - url = urljoin(obj.nexus_base_url, "/bank-connections/delete-connection") - try: - resp = post( - url, - json=dict(bankConnectionId=connection_name), - auth=auth.HTTPBasicAuth(obj.username, obj.password), - ) - except Exception: - print("Could not reach nexus at " + url) - exit(1) - - tell_user(resp) - - -@connections.command(help="restore backup") -@click.option("--backup-file", help="Back file", required=True) -@click.option("--passphrase", help="Passphrase for locking the backup", required=True) -@click.argument("connection-name") -@click.pass_obj -def restore_backup(obj, backup_file, passphrase, connection_name): - url = urljoin(obj.nexus_base_url, "/bank-connections") - try: - backup = open(backup_file, "r") - except Exception: - print("Could not open the backup at {}".format(backup_file)) - exit(1) - backup_json = json.loads(backup.read()) - backup.close() - - try: - resp = post( - url, - json=dict( - name=connection_name, - data=backup_json, - passphrase=passphrase, - source="backup", - ), - auth=auth.HTTPBasicAuth(obj.username, obj.password), - ) - except Exception: - print("Could not reach nexus at " + url) - exit(1) - - tell_user(resp) - - -@connections.command(help="make new EBICS bank connection") -@click.option("--ebics-url", help="EBICS URL", required=True) -@click.option("--host-id", help="Host ID", required=True) -@click.option("--partner-id", help="Partner ID", required=True) -@click.option("--ebics-user-id", help="Ebics user ID", required=True) -@click.argument("connection-name") -@click.pass_obj -def new_ebics_connection( - obj, connection_name, ebics_url, host_id, partner_id, ebics_user_id -): - url = urljoin(obj.nexus_base_url, "/bank-connections") - body = dict( - name=connection_name, - source="new", - type="ebics", - data=dict( - ebicsURL=ebics_url, - hostID=host_id, - partnerID=partner_id, - userID=ebics_user_id, - ), - ) - try: - resp = post(url, json=body, auth=auth.HTTPBasicAuth(obj.username, obj.password)) - except Exception: - print(f"Could not reach nexus at {url}") - exit(1) - - tell_user(resp) - - -@connections.command(help="Initialize the bank connection.") -@click.argument("connection-name") -@click.pass_obj -def connect(obj, connection_name): - url = urljoin(obj.nexus_base_url, f"/bank-connections/{connection_name}/connect") - try: - resp = post( - url, json=dict(), auth=auth.HTTPBasicAuth(obj.username, obj.password) - ) - except Exception: - print(f"Could not reach nexus at {url}") - exit(1) - tell_user(resp) - - -@connections.command(help="Import one bank account, chosen from the downloaded ones.") -@click.option( - "--offered-account-id", help="Name of the account to import", required=True -) -@click.option( - "--nexus-bank-account-id", - help="Name to give to the imported account", - required=True, -) -@click.argument("connection-name") -@click.pass_obj -def import_bank_account( - obj, connection_name, offered_account_id, nexus_bank_account_id -): - url = urljoin( - obj.nexus_base_url, - "/bank-connections/{}/import-account".format(connection_name), - ) - try: - resp = post( - url, - json=dict( - offeredAccountId=offered_account_id, - nexusBankAccountId=nexus_bank_account_id, - ), - auth=auth.HTTPBasicAuth(obj.username, obj.password), - ) - except Exception as e: - print(f"Could not reach nexus at {url}: {e}") - exit(1) - - tell_user(resp) - - -@connections.command(help="Update list of bank accounts available through this connection.") -@click.argument("connection-name") -@click.pass_obj -def download_bank_accounts(obj, connection_name): - url = urljoin( - obj.nexus_base_url, - "/bank-connections/{}/fetch-accounts".format(connection_name), - ) - try: - resp = post( - url, json=dict(), auth=auth.HTTPBasicAuth(obj.username, obj.password) - ) - except Exception: - print("Could not reach nexus at " + url) - exit(1) - - tell_user(resp) - - -@connections.command(help="List the connections.") -@click.pass_obj -def list_connections(obj): - url = urljoin(obj.nexus_base_url, "/bank-connections") - try: - resp = get( - url, json=dict(), auth=auth.HTTPBasicAuth(obj.username, obj.password) - ) - except Exception: - print("Could not reach nexus at " + url) - exit(1) - - tell_user(resp, withsuccess=True) - - -@connections.command(help="Show the status of a bank connection.") -@click.argument("connection-name") -@click.pass_obj -def show_connection(obj, connection_name): - url = urljoin(obj.nexus_base_url, f"/bank-connections/{connection_name}") - try: - resp = get( - url, json=dict(), auth=auth.HTTPBasicAuth(obj.username, obj.password) - ) - except Exception: - print("Could not reach nexus at " + url) - exit(1) - - tell_user(resp, withsuccess=True) - - -@connections.command(help="list bank accounts hosted at one connection") -@click.argument("connection-name") -@click.pass_obj -def list_offered_bank_accounts(obj, connection_name): - url = urljoin( - obj.nexus_base_url, "/bank-connections/{}/accounts".format(connection_name) - ) - try: - resp = get( - url, json=dict(), auth=auth.HTTPBasicAuth(obj.username, obj.password) - ) - except Exception: - print("Could not reach nexus at " + url) - exit(1) - - tell_user(resp, withsuccess=True) - - -@accounts.command(help="Schedules a new task") -@click.argument("account-name") -@click.option("--task-name", help="Name of the task", required=True) -@click.option("--task-cronspec", help="Cronspec string", required=True) -@click.option( - "--task-type", - help="'fetch' (downloads transactions histories) or 'submit' (uploads payments instructions)", - required=True, -) -@click.option( - "--task-param-range-type", - help="Only needed for 'fetch'. (FIXME: link to documentation here!)", - required=False, -) -@click.option( - "--task-param-level", - help="Only needed for 'fetch'. (FIXME: link to documentation here!)", - required=False, -) -@click.pass_obj -def task_schedule( - obj, - account_name, - task_name, - task_cronspec, - task_type, - task_param_range_type, - task_param_level, -): - - url = urljoin(obj.nexus_base_url, "/bank-accounts/{}/schedule".format(account_name)) - body = dict(name=task_name, cronspec=task_cronspec, type=task_type) - if task_type == "fetch" and not (task_param_range_type or task_param_level): - print("'fetch' type requires --task-param-range-type and --task-param-level") - return - - body.update( - dict(params=dict(rangeType=task_param_range_type, level=task_param_level)) - ) - try: - resp = post(url, json=body, auth=auth.HTTPBasicAuth(obj.username, obj.password)) - except Exception: - print("Could not reach nexus at " + url) - exit(1) - - tell_user(resp) - - -@accounts.command(help="Shows the status of one task") -@click.argument("account-name") -@click.option("--task-name", help="Name of the task", required=True) -@click.pass_obj -def task_status(obj, account_name, task_name): - url = urljoin( - obj.nexus_base_url, - "/bank-accounts/{}/schedule/{}".format(account_name, task_name), - ) - try: - resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password)) - except Exception: - print("Could not reach nexus " + url) - exit(1) - - tell_user(resp, withsuccess=True) - - -@accounts.command(help="Deletes one task") -@click.argument("account-name") -@click.option("--task-name", help="Name of the task", required=True) -@click.pass_obj -def task_delete(obj, account_name, task_name): - url = urljoin( - obj.nexus_base_url, - "/bank-accounts/{}/schedule/{}".format(account_name, task_name), - ) - try: - resp = delete(url, auth=auth.HTTPBasicAuth(obj.username, obj.password)) - except Exception: - print("Could not reach nexus " + url) - exit(1) - - tell_user(resp) - - -@accounts.command(help="Shows all the active tasks") -@click.argument("account-name") -@click.pass_obj -def tasks_show(obj, account_name): - url = urljoin(obj.nexus_base_url, "/bank-accounts/{}/schedule".format(account_name)) - try: - resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password)) - except Exception: - print("Could not reach nexus " + url) - exit(1) - - tell_user(resp, withsuccess=True) - - -@accounts.command(help="show accounts belonging to calling user") -@click.pass_obj -def show(obj): - url = urljoin(obj.nexus_base_url, "/bank-accounts") - try: - resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password)) - except Exception as e: - print(f"Could not reach nexus at {url}, error: {e}") - exit(1) - - tell_user(resp, withsuccess=True) - - -@accounts.command(help="prepare payment debiting 'account-name'") -@click.option( - "--creditor-iban", help="IBAN that will receive the payment", required=True -) -@click.option( - "--creditor-bic", help="BIC that will receive the payment", required=False -) -@click.option( - "--creditor-name", help="Legal name that will receive the payment", required=True -) -@click.option( - "--payment-amount", help="Amount to be paid (<currency>:X.Y)", required=True -) -@click.option("--payment-subject", help="Subject of this payment", required=True) -@click.argument("account-name") -@click.pass_obj -def prepare_payment( - obj, - account_name, - creditor_iban, - creditor_bic, - creditor_name, - payment_amount, - payment_subject, -): - url = urljoin( - obj.nexus_base_url, "/bank-accounts/{}/payment-initiations".format(account_name) - ) - body = dict( - iban=creditor_iban, - bic=creditor_bic, - name=creditor_name, - subject=payment_subject, - amount=payment_amount, - ) - - try: - resp = post(url, json=body, auth=auth.HTTPBasicAuth(obj.username, obj.password)) - except Exception: - print("Could not reach nexus at " + url) - exit(1) - - tell_user(resp, withsuccess=True) - - -@accounts.command(help="submit a prepared payment") -@click.option("--payment-uuid", help="payment unique identifier", required=True) -@click.argument("account-name") -@click.pass_obj -def submit_payment(obj, account_name, payment_uuid): - url = urljoin( - obj.nexus_base_url, - "/bank-accounts/{}/payment-initiations/{}/submit".format( - account_name, payment_uuid - ), - ) - try: - resp = post( - url, json=dict(), auth=auth.HTTPBasicAuth(obj.username, obj.password) - ) - except Exception: - print("Could not reach nexus at" + url) - exit(1) - - tell_user(resp) - - -@accounts.command(help="fetch transactions from the bank") -@click.option( - "--range-type", - default="all", - help="Admitted values: all, latest, previous-days, since-last", -) -@click.option("--level", default="all", help="Admitted values: report, statement, all") -@click.argument("account-name") -@click.pass_obj -def fetch_transactions(obj, account_name, range_type, level): - url = urljoin( - obj.nexus_base_url, "/bank-accounts/{}/fetch-transactions".format(account_name) - ) - try: - resp = post( - url, - json=dict(rangeType=range_type, level=level), - auth=auth.HTTPBasicAuth(obj.username, obj.password), - ) - except Exception: - print("Could not reach nexus " + url) - exit(1) - - tell_user(resp, withsuccess=True) - - -@accounts.command(help="get transactions from the simplified nexus JSON API") -@click.option( - "--compact/--no-compact", - help="Tells only amount/subject for each payment", - required=False, - default=False, -) -@click.argument("account-name") -@click.pass_obj -def transactions(obj, compact, account_name): - url = urljoin( - obj.nexus_base_url, "/bank-accounts/{}/transactions".format(account_name) - ) - try: - resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password)) - except Exception: - print("Could not reach nexus " + url) - exit(1) - - if compact and resp.status_code == 200: - for payment in resp.json()["transactions"]: - for entry in payment["batches"]: - for expected_singleton in entry["batchTransactions"]: - print( - "{}, {}".format( - expected_singleton["details"][ - "unstructuredRemittanceInformation" - ], - expected_singleton["amount"], - ) - ) - return - - tell_user(resp, withsuccess=True) - - -@facades.command(help="List active facades in the Nexus") -@click.argument("connection-name") -@click.pass_obj -def list_facades(obj, connection_name): - url = urljoin(obj.nexus_base_url, "/facades") - try: - resp = get(url, auth=auth.HTTPBasicAuth(obj.username, obj.password)) - except Exception as e: - print(f"Could not reach nexus (at {obj.nexus_base_url}): {e}") - exit(1) - - tell_user(resp, withsuccess=True) - - -@facades.command(help="create a new (Taler) facade") -@click.option("--facade-name", help="Name of the facade", required=True) -@click.option("--currency", help="Facade's currency", required=True) -@click.argument("connection-name") -@click.argument("account-name") -@click.pass_obj -def new_facade(obj, facade_name, connection_name, account_name, currency): - url = urljoin(obj.nexus_base_url, "/facades") - try: - resp = post( - url, - auth=auth.HTTPBasicAuth(obj.username, obj.password), - json=dict( - name=facade_name, - type="taler-wire-gateway", - config=dict( - currency=currency, - bankAccount=account_name, - bankConnection=connection_name, - reserveTransferLevel="UNUSED", - intervalIncremental="UNUSED", - ), - ), - ) - except Exception as e: - print(f"Could not reach nexus (at {obj.nexus_base_url}): {e}") - exit(1) - - tell_user(resp) - - -@sandbox.group("ebicshost", help="manage EBICS hosts") -@click.pass_context -def sandbox_ebicshost(ctx): - pass - - -@sandbox.command("check", help="check sandbox status") -@click.pass_obj -def check_sandbox_status(obj): - sandbox_base_url = obj.require_sandbox_base_url() - url = urljoin(sandbox_base_url, "/config") - try: - resp = get(url) - except Exception: - print("Could not reach sandbox") - exit(1) - - tell_user(resp, withsuccess=True) - - -@sandbox_ebicshost.command("create", help="Create an EBICS host") -@click.option("--host-id", help="EBICS host ID", required=True, prompt=True) -@click.pass_obj -def make_ebics_host(obj, host_id): - sandbox_base_url = obj.require_sandbox_base_url() - url = urljoin(sandbox_base_url, "/admin/ebics/hosts") - try: - resp = post(url, json=dict(hostID=host_id, ebicsVersion="2.5")) - except Exception: - print("Could not reach sandbox") - exit(1) - - tell_user(resp) - - -@sandbox_ebicshost.command("list", help="List EBICS hosts.") -@click.pass_obj -def list_ebics_host(obj): - sandbox_base_url = obj.require_sandbox_base_url() - url = urljoin(sandbox_base_url, "/admin/ebics/hosts") - try: - resp = get(url) - except Exception: - print("Could not reach sandbox") - exit(1) - - tell_user(resp, withsuccess=True) - - -@sandbox.group("ebicssubscriber", help="manage EBICS subscribers") -@click.pass_context -def sandbox_ebicssubscriber(ctx): - pass - - -@sandbox_ebicssubscriber.command("create", help="Create an EBICS subscriber.") -@click.option("--host-id", help="Ebics host ID", required=True, prompt=True) -@click.option("--partner-id", help="Ebics partner ID", required=True, prompt=True) -@click.option("--user-id", help="Ebics user ID", required=True, prompt=True) -@click.pass_obj -def create_ebics_subscriber(obj, host_id, partner_id, user_id): - sandbox_base_url = obj.require_sandbox_base_url() - url = urljoin(sandbox_base_url, "/admin/ebics/subscribers") - try: - resp = post( - url, json=dict(hostID=host_id, partnerID=partner_id, userID=user_id) - ) - except Exception: - print("Could not reach sandbox") - exit(1) - - tell_user(resp) - - -@sandbox_ebicssubscriber.command("list", help="List EBICS subscribers.") -@click.pass_obj -def list_ebics_subscriber(obj): - sandbox_base_url = obj.require_sandbox_base_url() - url = urljoin(sandbox_base_url, "/admin/ebics/subscribers") - try: - resp = get(url) - except Exception: - print("Could not reach sandbox") - exit(1) - - tell_user(resp, withsuccess=True) - - -@sandbox.group("ebicsbankaccount", help="manage EBICS bank accounts") -@click.pass_context -def sandbox_ebicsbankaccount(ctx): - pass - - -@sandbox_ebicsbankaccount.command( - "create", help="Create a bank account for a EBICS subscriber." -) -@click.option("--currency", help="currency", prompt=True) -@click.option("--iban", help="IBAN", required=True) -@click.option("--bic", help="BIC", required=True) -@click.option("--person-name", help="bank account owner name", required=True) -@click.option("--account-name", help="label of this bank account", required=True) -@click.option("--ebics-user-id", help="user ID of the Ebics subscriber", required=True) -@click.option("--ebics-host-id", help="host ID of the Ebics subscriber", required=True) -@click.option( - "--ebics-partner-id", help="partner ID of the Ebics subscriber", required=True -) -@click.pass_obj -def associate_bank_account( - obj, - currency, - iban, - bic, - person_name, - account_name, - ebics_user_id, - ebics_host_id, - ebics_partner_id, -): - sandbox_base_url = obj.require_sandbox_base_url() - url = urljoin(sandbox_base_url, "/admin/ebics/bank-accounts") - body = dict( - currency=currency, - subscriber=dict( - userID=ebics_user_id, partnerID=ebics_partner_id, hostID=ebics_host_id - ), - iban=iban, - bic=bic, - name=person_name, - label=account_name, - ) - - try: - resp = post(url, json=body) - except Exception: - print("Could not reach sandbox") - exit(1) - - tell_user(resp) - - -@sandbox.group("bankaccount", help="manage bank accounts") -@click.pass_context -def sandbox_bankaccount(ctx): - pass - - -@sandbox_bankaccount.command("list", help="List accounts") -@click.pass_obj -def bankaccount_list(obj): - sandbox_base_url = obj.require_sandbox_base_url() - url = urljoin(sandbox_base_url, f"/admin/bank-accounts") - try: - resp = get(url) - except Exception: - print("Could not reach sandbox") - exit(1) - - tell_user(resp, withsuccess=True) - - -@sandbox_bankaccount.command("transactions", help="List transactions") -@click.argument("account-label") -@click.pass_obj -def transactions_list(obj, account_label): - sandbox_base_url = obj.require_sandbox_base_url() - url = urljoin( - sandbox_base_url, f"/admin/bank-accounts/{account_label}/transactions" - ) - try: - resp = get(url) - except Exception: - print("Could not reach sandbox") - exit(1) - - tell_user(resp, withsuccess=True) - - -@sandbox_bankaccount.command("generate-transactions", help="Generate test transactions") -@click.argument("account-label") -@click.pass_obj -def bankaccount_generate_transactions(obj, account_label): - sandbox_base_url = obj.require_sandbox_base_url() - url = urljoin( - sandbox_base_url, f"/admin/bank-accounts/{account_label}/generate-transactions" - ) - try: - resp = post(url) - except Exception: - print("Could not reach sandbox") - exit(1) - - tell_user(resp) - - -@sandbox_bankaccount.command(help="Book a payment in the sandbox") -@click.option("--creditor-iban", help="IBAN receiving the payment", prompt=True) -@click.option("--creditor-bic", help="BIC receiving the payment", prompt=True) -@click.option( - "--creditor-name", - help="Name of the person who is receiving the payment", - prompt=True, -) -@click.option("--debtor-iban", help="IBAN sending the payment", prompt=True) -@click.option("--debtor-bic", help="BIC sending the payment", prompt=True) -@click.option( - "--debtor-name", help="name of the person who is sending the payment", prompt=True -) -@click.option("--amount", help="amount, no currency", prompt=True) -@click.option("--currency", help="currency", prompt=True) -@click.option("--subject", help="payment subject", prompt=True) -@click.option( - "--direction", - help="direction respect to the bank account hosted at Sandbox: allows DBIT/CRDT values.", - prompt=True, -) -@click.pass_obj -def book_payment( - obj, - creditor_iban, - creditor_bic, - creditor_name, - debtor_iban, - debtor_bic, - debtor_name, - amount, - currency, - subject, - direction, -): - sandbox_base_url = obj.require_sandbox_base_url() - url = urljoin(sandbox_base_url, "/admin/payments") - body = dict( - creditorIban=creditor_iban, - creditorBic=creditor_bic, - creditorName=creditor_name, - debitorIban=debtor_iban, - debitorBic=debtor_bic, - debitorName=debtor_name, - amount=amount, - currency=currency, - subject=subject, - direction=direction, - ) - try: - resp = post(url, json=body) - except Exception: - print("Could not reach sandbox") - exit(1) - - tell_user(resp) - - -cli(obj={}) |