summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMS <ms@taler.net>2021-01-21 16:44:15 +0100
committerMS <ms@taler.net>2021-01-21 16:44:15 +0100
commit80478f3885f2b3c56ed7d7b0127784549d75c7d2 (patch)
tree8cb7fecbcb5643328a41ceac6b7f348e2b487d1c
parentdf7d8c253147b83cfcfbc962c1a244f8fe8060bf (diff)
downloadlibeufin-80478f3885f2b3c56ed7d7b0127784549d75c7d2.tar.gz
libeufin-80478f3885f2b3c56ed7d7b0127784549d75c7d2.tar.bz2
libeufin-80478f3885f2b3c56ed7d7b0127784549d75c7d2.zip
remove unwanted
-rw-r--r--build.gradle10
-rwxr-xr-xcli/bin/libeufin-cli/libeufin-cli.template1048
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={})