exchange/contrib/taler-nexus-prepare

#!/usr/bin/env python3
# This file is in the public domain.

from requests import get, post
from subprocess import call
import base64

# EBICS details
EBICS_URL = "http://localhost:5000/ebicsweb"
HOST_ID = "HOST01"
PARTNER_ID = "PARTNER1"
USER_ID = "USER1"
EBICS_VERSION = "H004"

SUBSCRIBER_IBAN = "ES9121000418450200051332"
SUBSCRIBER_BIC = "BIC"
SUBSCRIBER_NAME = "Exchange"

BANK_ACCOUNT_LABEL = "my-bank-account"
BANK_CONNECTION_LABEL = "my-bank-connection"
FACADE_LABEL="my-facade"

USERNAME="Exchange"
USER_AUTHORIZATION_HEADER = "basic {}".format(
    base64.b64encode(b"Exchange:x").decode("utf-8")
)

def assertResponse(response):
    if response.status_code != 200:
        print("Test failed on URL: {}".format(response.url))
        # stdout/stderr from both services is A LOT of text.
        # Confusing to dump all that to console.
        print("Check nexus.log and sandbox.log, probably under /tmp")
        exit(1)
    # Allows for finer grained checks.
    return response

# Create a nexus (super-) user
check_call(["libeufin-nexus",
      "superuser",
      "--db-name", "/tmp/nexus-exchange-test.sqlite3",
      "Exchange",
      "--password", "x"]
)

# Create a EBICS bank connection.
assertResponse(
    post(
        "http://localhost:5001/bank-connections",
        json=dict(
            name=BANK_CONNECTION_LABEL,
            source="new",
            type="ebics",
            data=dict(
                ebicsURL=EBICS_URL, hostID=HOST_ID, partnerID=PARTNER_ID, userID=USER_ID
            ),
        ),
        headers=dict(Authorization=USER_AUTHORIZATION_HEADER),
    )
)

# Create a facade
assertResponse(
    post(
        "http://localhost:5001/facades",
        json=dict(
            name=FACADE_LABEL,
            type="taler-wire-gateway",
            creator=USERNAME,
            config=dict(
                bankAccount=BANK_ACCOUNT_LABEL,
                bankConnection=BANK_CONNECTION_LABEL,
                reserveTransferLevel="UNUSED",
                intervalIncremental="UNUSED"
            )
        ),
        headers=dict(Authorization=USER_AUTHORIZATION_HEADER),
    )
)

# Create the EBICS host at the Sandbox.
assertResponse(
    post(
        "http://localhost:5000/admin/ebics/host",
        json=dict(hostID=HOST_ID, ebicsVersion=EBICS_VERSION),
    )
)

# Create Exchange EBICS subscriber at the Sandbox.
assertResponse(
    post(
        "http://localhost:5000/admin/ebics/subscribers",
        json=dict(hostID=HOST_ID, partnerID=PARTNER_ID, userID=USER_ID),
    )
)

# Create a bank account associated to the Exchange's EBICS subscriber,
# again at the Sandbox.
assertResponse(
    post(
        "http://localhost:5000/admin/ebics/bank-accounts",
        json=dict(
            subscriber=dict(hostID=HOST_ID, partnerID=PARTNER_ID, userID=USER_ID),
            iban=SUBSCRIBER_IBAN,
            bic=SUBSCRIBER_BIC,
            name=SUBSCRIBER_NAME,
            label=BANK_ACCOUNT_LABEL,
        ),
    )
)

# 'connect' to the bank: upload+download keys.
assertResponse(
    post(
        "http://localhost:5001/bank-connections/{}/connect".format(BANK_CONNECTION_LABEL),
        json=dict(),
        headers=dict(Authorization=USER_AUTHORIZATION_HEADER),
    )
)

# Download bank accounts.
assertResponse(
    post(
        "http://localhost:5001/bank-connections/{}/ebics/import-accounts".format(BANK_CONNECTION_LABEL),
        json=dict(),
        headers=dict(Authorization=USER_AUTHORIZATION_HEADER),
    )
)