[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[libeufin] branch master updated: re-add CLI
|
From: |
gnunet |
|
Subject: |
[libeufin] branch master updated: re-add CLI |
|
Date: |
Sat, 23 Jan 2021 16:56:50 +0100 |
This is an automated email from the git hooks/post-receive script.
dold pushed a commit to branch master
in repository libeufin.
The following commit(s) were added to refs/heads/master by this push:
new d29286e re-add CLI
d29286e is described below
commit d29286e4601f318c68555b9b78b76394cdbe7202
Author: Florian Dold <florian@dold.me>
AuthorDate: Sat Jan 23 16:56:43 2021 +0100
re-add CLI
---
.gitignore | 1 -
build.gradle | 4 +-
cli/bin/libeufin-cli/libeufin-cli.template | 1048 ++++++++++++++++++++++++++++
3 files changed, 1050 insertions(+), 3 deletions(-)
diff --git a/.gitignore b/.gitignore
index f554aff..c943cec 100644
--- a/.gitignore
+++ b/.gitignore
@@ -19,4 +19,3 @@ __pycache__
.DS_Store
*.mk
util/src/main/resources/version.txt
-cli/bin/libeufin-cli
diff --git a/build.gradle b/build.gradle
index 236ea3f..c7fb639 100644
--- a/build.gradle
+++ b/build.gradle
@@ -32,7 +32,7 @@ task versionFile() {
task replaceVersionCli(type: Copy) {
from file("cli/bin/libeufin-cli")
- into file("$project.buildDir/generated/python")
+ into file("${project.buildDir}/generated/python")
filter(ReplaceTokens, tokens: [version: getRootProject().version])
}
@@ -62,7 +62,7 @@ task dist(type: Zip) {
include("**/libeufin-sandbox")
include("**/*.jar")
}
- from("$project.buildDir/generated/python") {
+ from("${project.buildDir}/generated/python") {
include("libeufin-cli")
rename { "bin/libeufin-cli" }
}
diff --git a/cli/bin/libeufin-cli/libeufin-cli.template
b/cli/bin/libeufin-cli/libeufin-cli.template
new file mode 100755
index 0000000..c7aabe5
--- /dev/null
+++ b/cli/bin/libeufin-cli/libeufin-cli.template
@@ -0,0 +1,1048 @@
+#!/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={})
--
To stop receiving notification emails like this one, please contact
gnunet@gnunet.org.
| [Prev in Thread] |
Current Thread |
[Next in Thread] |
- [libeufin] branch master updated: re-add CLI,
gnunet <=