diff options
Diffstat (limited to 'integration-tests')
-rw-r--r-- | integration-tests/json_checks.py | 131 | ||||
-rwxr-xr-x | integration-tests/tests.py | 554 | ||||
-rw-r--r-- | integration-tests/util.py | 211 |
3 files changed, 0 insertions, 896 deletions
diff --git a/integration-tests/json_checks.py b/integration-tests/json_checks.py deleted file mode 100644 index 916712a6..00000000 --- a/integration-tests/json_checks.py +++ /dev/null @@ -1,131 +0,0 @@ -#!/usr/bin/env python3 - -# This minimal library checks only if the JSON values -# contains the expected fields, without actually checking -# if the fields' value match the API. - -from util import CheckJsonField as F, CheckJsonTop as T - -def checkNewUserRequest(json): - c = T(F("username"), F("password")) - return c.check(json) - -def checkBankAccountElement(json): - c = T( - F("nexusBankAccountId"), - F("iban"), - F("bic"), - F("ownerName") - ) - return c.check(json) - -def checkPreparePayment(json): - c = T( - F("iban"), - F("bic"), - F("name"), - F("subject"), - F("amount") - ) - return c.check(json) - -def checkBankConnection(json): - c = T( - F("bankConnectionId"), - F("bankConnectionType"), - F("ready"), - F("bankKeysReviewed") - ) - return c.check(json) - -def checkDeleteConnection(json): - c = T(F("bankConnectionId")) - return c.check(json) - -def checkConnectionListElement(json): - c = T( - F("name"), - F("type") - ) - return c.check(json) - -def checkPreparedPaymentResponse(json): - c = T(F("uuid")) - return c.check(json) - -def checkPreparedPaymentElement(json): - c = T( - F("paymentInitiationId"), - F("submitted"), - F("creditorIban"), - F("creditorBic"), - F("creditorName"), - F("amount"), - F("subject"), - F("submissionDate"), - F("preparationDate") - ) - return c.check(json) - -def checkFetchTransactions(json): - c = T( - F("rangeType"), - F("level") - ) - return c.check(json) - -def checkTransaction(json): - c = T( - F("account"), - F("counterpartIban"), - F("counterpartBic"), - F("counterpartName"), - F("amount"), - F("date"), - F("subject") - ) - return c.check(json) - -def checkNewEbicsConnection(json): - c = T( - F("source"), - F("name"), - F("type"), - F("data", T( - F("ebicsURL"), - F("hostID"), - F("partnerID"), - F("userID"), - )) - ) - return c.check(json) - -def checkImportAccount(json): - c = T(F("nexusBankAccountId"), - F("offeredAccountId")) - return c.check(json) - -def checkSandboxEbicsHosts(json): - c = T(F("hostID"), - F("ebicsVersion")) - return c.check(json) - -def checkSandboxEbicsSubscriber(json): - c = T(F("hostID"), - F("userID"), - F("partnerID"), - F("systemID", optional=True)) - return c.check(json) - -def checkSandboxBankAccount(json): - c = T( - F("iban"), - F("bic"), - F("name"), - F("subscriber"), - F("label") - ) - return c.check(json) - -def checkBackupDetails(json): - return T(F("passphrase")).check(json) diff --git a/integration-tests/tests.py b/integration-tests/tests.py deleted file mode 100755 index ab7720bd..00000000 --- a/integration-tests/tests.py +++ /dev/null @@ -1,554 +0,0 @@ -#!/usr/bin/env python3 - -import os -import pytest -import json -from deepdiff import DeepDiff as dd -from subprocess import check_call -from requests import post, get, auth, delete -from time import sleep -from util import ( - startNexus, - startSandbox, - assertResponse, - makeNexusSuperuser, - dropSandboxTables, - dropNexusTables, - assertJsonEqual, - LibeufinPersona, - BankingDetails, - NexusDetails, - EbicsDetails, - compileLibeufin -) - -# Database -DB = "jdbc:sqlite:/tmp/libeufintestdb" -SANDBOX_URL = "http://localhost:5000" -NEXUS_URL = "http://localhost:5001" - -user0 = LibeufinPersona( - banking_details = BankingDetails(SANDBOX_URL), - nexus_details = NexusDetails(NEXUS_URL), - ebics_details = EbicsDetails(SANDBOX_URL + "/ebicsweb") -) - -os.environ["LIBEUFIN_NEXUS_DB_CONNECTION"] = DB -os.environ["LIBEUFIN_SANDBOX_DB_CONNECTION"] = DB - -def prepareSandbox(): - # make ebics host at sandbox - assertResponse( - post( - f"{user0.banking.bank_base_url}/admin/ebics/hosts", - json=dict( - hostID=user0.ebics.host, - ebicsVersion=user0.ebics.version - ), - ) - ) - # make new ebics subscriber at sandbox - assertResponse( - post( - f"{user0.banking.bank_base_url}/admin/ebics/subscribers", - json=user0.ebics.get_as_dict(with_url=False), - ) - ) - # give a bank account to such subscriber, at sandbox - assertResponse( - post( - f"{user0.banking.bank_base_url}/admin/ebics/bank-accounts", - json=dict( - name=user0.banking.name, - subscriber=user0.ebics.get_as_dict(with_url=False), - iban=user0.banking.iban, - bic=user0.banking.bic, - label=user0.banking.label, - currency="EUR" - ) - ) - ) - -# most of the operations are run by the superuser. -def prepareNexus(): - makeNexusSuperuser() - assertResponse( - post( - f"{user0.nexus.base_url}/bank-connections", - json=dict( - name=user0.nexus.bank_connection, - source="new", - type="ebics", - data=user0.ebics.get_as_dict(with_url=True), - ), - auth=user0.nexus.auth - ) - ) - # synchronizing the connection - assertResponse( - post( - f"{user0.nexus.base_url}/bank-connections/{user0.nexus.bank_connection}/connect", - json=dict(), - auth=user0.nexus.auth - ) - ) - # download offered bank accounts - assertResponse( - post( - f"{user0.nexus.base_url}/bank-connections/{user0.nexus.bank_connection}/fetch-accounts", - json=dict(), - auth=user0.nexus.auth - ) - ) - # import one bank account into the Nexus - assertResponse( - post( - f"{user0.nexus.base_url}/bank-connections/{user0.nexus.bank_connection}/import-account", - json=dict( - offeredAccountId=user0.banking.label, - nexusBankAccountId=user0.nexus.bank_label - ), - auth=user0.nexus.auth - ) - ) - -compileLibeufin() -dropSandboxTables() -startSandbox() -dropNexusTables() -startNexus() - -def setup_function(): - try: - prepareSandbox() - prepareNexus() - except Exception: - teardown_function() - assert False, "Setup function failed" - -def teardown_function(): - dropSandboxTables() - dropNexusTables() - -def test_double_username(): - assertResponse( - post(f"{user0.nexus.base_url}/users", - auth=user0.nexus.auth, - json=dict(username="admin", password="secret") - ), - acceptedResponses=[409] - ) - -def test_change_nonadmin_password(): - assertResponse( - post(f"{user0.nexus.base_url}/users", - auth=user0.nexus.auth, - json=dict(username="nonadmin", password="secret") - ) - ) - - resp = assertResponse( - get( - f"{user0.nexus.base_url}/bank-accounts", - auth=auth.HTTPBasicAuth("nonadmin", "secret") - ) - ) - - assertResponse( - post(f"{user0.nexus.base_url}/users/password", - auth=auth.HTTPBasicAuth("nonadmin", "secret"), - json=dict(newPassword="changed") - ) - ) - - assertResponse( - get( - f"{user0.nexus.base_url}/bank-accounts", - auth=auth.HTTPBasicAuth("nonadmin", "changed") - ) - ) - - -def test_connection_deletion(): - resp = assertResponse( - get( - f"{user0.nexus.base_url}/bank-connections", - auth=user0.nexus.auth - ) - ) - connection = resp.json().get("bankConnections").pop() - assert user0.nexus.bank_connection == connection.get("name") - - assertResponse( - post( - f"{user0.nexus.base_url}/bank-connections/delete-connection", - json=dict(bankConnectionId=user0.nexus.bank_connection), - auth=user0.nexus.auth - ) - ) - resp = assertResponse( - get( - f"{user0.nexus.base_url}/bank-connections", - auth=user0.nexus.auth - ) - ) - assert len(resp.json().get("bankConnections")) == 0 - -# Tests whether Nexus knows the imported bank account. -def test_imported_account(): - resp = assertResponse( - get( - f"{user0.nexus.base_url}/bank-connections/{user0.nexus.bank_connection}/accounts", - auth=user0.nexus.auth - ) - ) - imported_account = resp.json().get("accounts").pop() - assert imported_account.get("nexusBankAccountId") == user0.nexus.bank_label - -# Expecting a empty history for an account that -# never made or receivd a payment. -def test_empty_history(): - resp = assertResponse( - get( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/transactions", - auth=user0.nexus.auth - ) - ) - assert len(resp.json().get("transactions")) == 0 - -# This test checks the bank connection backup export+import -# However the restored connection is never sent through the -# "/connect" step. -def test_backup(): - resp = assertResponse( - post( - f"{user0.nexus.base_url}/bank-connections/{user0.nexus.bank_connection}/export-backup", - json=dict(passphrase="secret"), - auth=user0.nexus.auth - ) - ) - sleep(3) - assertResponse( - post( - f"{user0.nexus.base_url}/bank-connections", - json=dict(name="my-ebics-restored", data=resp.json(), passphrase="secret", source="backup"), - auth=user0.nexus.auth - ) - ) - -def test_ebics_custom_ebics_order(): - assertResponse( - post( - f"{user0.nexus.base_url}/bank-connections/{user0.nexus.bank_connection}/ebics/download/tsd", - auth=user0.nexus.auth - ) - ) - -# This test makes a payment and expects to see it -# in the account history. -def test_payment(): - resp = assertResponse(post( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/payment-initiations", - json=dict( - iban="FR7630006000011234567890189", - bic="AGRIFRPP", - name="Jacques La Fayette", - subject="integration test", - amount="EUR:1", - ), - auth=user0.nexus.auth - )) - PAYMENT_UUID = resp.json().get("uuid") - assertResponse(post("/".join([ - user0.nexus.base_url, - "bank-accounts", - user0.nexus.bank_label, - "payment-initiations", - PAYMENT_UUID, - "submit"]), - json=dict(), - auth=user0.nexus.auth - )) - assertResponse( - post( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/fetch-transactions", - auth=user0.nexus.auth - ) - ) - resp = assertResponse(get( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/transactions", - auth=user0.nexus.auth - )) - assert len(resp.json().get("transactions")) == 1 - - # assert now that the bank booked the payment. - resp = assertResponse(get("/".join([ - user0.nexus.base_url, - "bank-accounts", - user0.nexus.bank_label, - "payment-initiations", - PAYMENT_UUID]), - auth=user0.nexus.auth - )) - assert resp.json()["status"] == "BOOK" - - # Posting a second payment initiation, but not submitting it. - # It's expected to have a "null" status. - resp = assertResponse(post( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/payment-initiations", - json=dict( - iban="FR7630006000011234567890189", - bic="AGRIFRPP", - name="Jacques La Fayette", - subject="integration test", - amount="EUR:1", - ), - auth=user0.nexus.auth - )) - PAYMENT_UUID_NON_SUBMITTED = resp.json().get("uuid") - resp = assertResponse(get("/".join([ - user0.nexus.base_url, - "bank-accounts", - user0.nexus.bank_label, - "payment-initiations", - PAYMENT_UUID_NON_SUBMITTED]), - auth=user0.nexus.auth - )) - assert resp.json()["status"] == None - - -@pytest.fixture -def fetch_transactions(): - assertResponse(post( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/fetch-transactions", - auth=user0.nexus.auth - )) - -# Book a incoming payment for "persona" at the Sandbox. -@pytest.fixture -def make_crdt_payment(): - payment_instruction = dict( - creditorIban=user0.banking.iban, - creditorBic=user0.banking.bic, - creditorName=user0.banking.name, - debitorIban="FR00000000000000000000", - debitorBic="BUKBGB22", - debitorName="Max Mustermann", - amount=5, - uid="AcctSvcrRef-56789", - currency="EUR", - subject="Reimbursement", - direction="CRDT" - ) - - assertResponse(post( - f"{user0.banking.bank_base_url}/admin/payments", - json=payment_instruction - )) - - -def test_deduplication(make_crdt_payment): - # fetching twice the transactions and check that - # the payment made via the fixture shows up only once. - assertResponse(post( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/fetch-transactions", - auth=user0.nexus.auth - )) - assertResponse(post( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/fetch-transactions", - auth=user0.nexus.auth - )) - resp = assertResponse(get( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/transactions", - auth=user0.nexus.auth - )) - assert len(resp.json().get("transactions")) == 1 - - -def test_facade_name_collision(): - assertResponse( - post( - f"{user0.nexus.base_url}/facades", - json=dict( - name=user0.nexus.taler_facade_name, - type="taler-wire-gateway", - config=dict( - currency="EUR", - bankAccount=user0.nexus.bank_label, - bankConnection=user0.nexus.bank_connection, - reserveTransferLevel="UNUSED", - ) - ), - auth=user0.nexus.auth - ) - ) - assertResponse( - post( - f"{user0.nexus.base_url}/facades", - json=dict( - name=user0.nexus.taler_facade_name, - type="taler-wire-gateway", - config=dict( - currency="EUR", - bankAccount=user0.nexus.bank_label, - bankConnection=user0.nexus.bank_connection, - reserveTransferLevel="UNUSED", - ) - ), - auth=user0.nexus.auth - ), - acceptedResponses=[400] - ) - -def test_double_connection_name(): - assertResponse( - post( - f"{user0.nexus.base_url}/bank-connections", - json=dict( - name=user0.nexus.bank_connection, - source="new", - type="ebics", - data=user0.ebics.get_as_dict(with_url=True), - ), - auth=user0.nexus.auth - ), - [409] # Conflict - ) - -def test_ingestion_camt53_non_singleton(): - with open("../parsing-tests/samples/camt53-gls-style-1.xml") as f: - camt = f.read() - assertResponse( - post( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/test-camt-ingestion/C53", - auth=user0.nexus.auth, - data=camt - ) - ) - resp = assertResponse( - get( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/transactions", - auth=user0.nexus.auth - ) - ) - with open("../parsing-tests/samples/camt53-gls-style-1.json") as f: - expected_txs = f.read() - assert not dd(resp.json(), json.loads(expected_txs), ignore_order=True) - - -def test_ingestion_camt53(): - with open("../parsing-tests/samples/camt53-gls-style-0.xml") as f: - camt = f.read() - assertResponse( - post( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/test-camt-ingestion/C53", - auth=user0.nexus.auth, - data=camt - ) - ) - resp = assertResponse( - get( - f"{user0.nexus.base_url}/bank-accounts/{user0.nexus.bank_label}/transactions", - auth=user0.nexus.auth - ) - ) - with open("../parsing-tests/samples/camt53-gls-style-0.json") as f: - expected_txs = f.read() - assert not dd(resp.json(), json.loads(expected_txs), ignore_order=True) - -def test_sandbox_camt(): - payment_instruction = dict( - creditorIban="GB33BUKB20201555555555", - creditorBic="BUKBGB22", - creditorName="Oliver Smith", - debitorIban="FR00000000000000000000", - debitorBic="BUKBGB22", - debitorName="Max Mustermann", - amount=5, - currency="EUR", - subject="Reimbursement", - direction="CRDT", - uid="AcctSvcrRef-01234" - ) - assertResponse( - post( - f"{user0.banking.bank_base_url}/admin/payments", - json=payment_instruction - ) - ) - - assertResponse( - post( - f"{user0.banking.bank_base_url}/admin/payments/camt", - json=dict(iban="GB33BUKB20201555555555", type=53) - ) - ) - -def test_schedule_deletion(): - assertResponse( - post("/".join([ - user0.nexus.base_url, - "bank-accounts", - user0.nexus.bank_label, - "schedule"]), - json=dict( - name="test-task", - cronspec="* * *", - type="fetch", - params=dict(rangeType="all", level="all")), - auth=auth.HTTPBasicAuth("admin", "x") - ) - ) - - resp = assertResponse( - get("/".join([ - user0.nexus.base_url, - "bank-accounts", - user0.nexus.bank_label, - "schedule", - "test-task"]), - auth=auth.HTTPBasicAuth("admin", "x") - ) - ) - assert resp.json().get("taskName") == "test-task" - - assertResponse( - delete("/".join([ - user0.nexus.base_url, - "bank-accounts", - user0.nexus.bank_label, - "schedule", - "test-task"]), - auth=auth.HTTPBasicAuth("admin", "x") - ) - ) - - resp = assertResponse( - get("/".join([ - user0.nexus.base_url, - "bank-accounts", - user0.nexus.bank_label, - "schedule", - "test-task"]), - auth=auth.HTTPBasicAuth("admin", "x") - ), - acceptedResponses=[404] - ) - -def test_invalid_json(): - assertResponse( - post( - f"{user0.nexus.base_url}/users", - data="malformed", - headers={"Content-Type": "application/json"}, - auth=auth.HTTPBasicAuth("admin", "x")), - acceptedResponses=[400] - ) - assertResponse( - post( - f"{user0.banking.bank_base_url}/admin/ebics/hosts", - data="malformed", - headers={"Content-Type": "application/json"}, - auth=auth.HTTPBasicAuth("admin", "x")), - acceptedResponses=[400] - ) diff --git a/integration-tests/util.py b/integration-tests/util.py deleted file mode 100644 index 49606115..00000000 --- a/integration-tests/util.py +++ /dev/null @@ -1,211 +0,0 @@ -# Helpers for the integration tests. - -from subprocess import check_call, Popen, PIPE, DEVNULL -import socket -from requests import post, get, auth -from time import sleep -from deepdiff import DeepDiff -import atexit -from pathlib import Path -import sys -import os - -class EbicsDetails: - def get_as_dict(self, with_url): - ret = dict( - hostID=self.host, - partnerID=self.partner, - userID=self.user - ) - if with_url: - ret.update(ebicsURL=self.service_url) - return ret - - def __init__(self, service_url): - self.service_url = service_url - self.host = "HOST01" - self.partner = "PARTNER1" - self.user = "USER1" - self.version = "H004" - -class BankingDetails: - def __init__( - self, - bank_base_url, - iban="GB33BUKB20201555555555", - bic="BUKBGB22", - label="savings", - name="Oliver Smith" - ): - self.iban = iban - self.bic = bic - self.label = label - self.bank_base_url = bank_base_url - self.name = name - -class NexusDetails: - def __init__( - self, base_url, - username = "admin", - password = "x", - bank_connection_name = "my-ebics", - bank_local_name = "local-savings"): - self.base_url = base_url - self.username = username - self.password = password - self.bank_connection = bank_connection_name - self.bank_label = bank_local_name - self.auth = auth.HTTPBasicAuth(self.username, self.password) - self.taler_facade_name = "taler-wire-gateway" - -class LibeufinPersona: - def __init__(self, banking_details, nexus_details, ebics_details): - self.banking = banking_details - self.nexus = nexus_details - self.ebics = ebics_details - -class CheckJsonField: - def __init__(self, name, nested=None, optional=False): - self.name = name - self.nested = nested - self.optional = optional - - def check(self, json): - if self.name not in json and not self.optional: - print(f"'{self.name}' not found in the JSON: {json}.") - sys.exit(1) - if self.nested: - self.nested.check(json.get(self.name)) - -class CheckJsonTop: - def __init__(self, *args): - self.checks = args - - def check(self, json): - for check in self.checks: - check.check(json) - return json - - -def assertJsonEqual(json1, json2): - diff = DeepDiff(json1, json2, ignore_order=True, report_repetition=True) - assert len(diff.keys()) == 0 - - -def checkPort(port): - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - try: - s.bind(("0.0.0.0", port)) - s.close() - except: - print(f"Port {port} is not available") - print(sys.exc_info()[0]) - exit(77) - -def kill(name, s): - s.terminate() - s.wait() - -def makeNexusSuperuser(): - check_call([ - "../gradlew", - "-q", "--console=plain", - "-p", "..", - "nexus:run", - f"--args=superuser admin --password x", - ]) - -def dropSandboxTables(): - check_call([ - "../gradlew", - "-q", "--console=plain", - "-p", "..", - "sandbox:run", - f"--args=reset-tables" - ]) - - -def compileLibeufin(): - check_call([ - "../gradlew", - "-q", "--console=plain", - "-p", "..", - "assemble" - ]) - -def dropNexusTables(): - check_call([ - "../gradlew", - "-q", "--console=plain", - "-p", "..", - "nexus:run", - f"--args=reset-tables" - ]) - -def startSandbox(): - checkPort(5000) - sandbox = Popen([ - "../gradlew", - "-q", - "-p", - "..", - "sandbox:run", - "--console=plain", - "--args=serve"], - stdin=DEVNULL, - stdout=open("sandbox-stdout.log", "w"), - stderr=open("sandbox-stderr.log", "w") - ) - atexit.register(lambda: kill("sandbox", sandbox)) - for i in range(10): - try: - get("http://localhost:5000/") - except: - if i == 9: - stdout, stderr = sandbox.communicate() - print("Sandbox timed out") - print("{}\n{}".format(stdout.decode(), stderr.decode())) - exit(77) - sleep(2) - continue - break - - -def startNexus(): - checkPort(5001) - nexus = Popen([ - "../gradlew", - "-q", - "-p", - "..", - "nexus:run", - "--console=plain", - "--args=serve"], - stdin=DEVNULL, - stdout=open("nexus-stdout.log", "w"), - stderr=open("nexus-stderr.log", "w") - ) - atexit.register(lambda: kill("nexus", nexus)) - for i in range(80): - try: - get("http://localhost:5001/") - except: - if i == 79: - nexus.terminate() - print("Nexus timed out") - exit(77) - sleep(1) - continue - break - return nexus - -def assertResponse(r, acceptedResponses=[200]): - def http_trace(r): - request = f"{r.request.method} {r.request.url}" - if r.request.body: - request += f"\n{r.request.body.decode('utf-8')}" - response = f"{r.status_code} {r.reason}\n{r.text}" - return f"(the following communication failed)\n\n{request}\n\n{response}" - assert r.status_code in acceptedResponses, http_trace(r) - return r |