gnunet-svn
[Top][All Lists]
Advanced

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

[libeufin] branch master updated: Dropping the integration tests.


From: gnunet
Subject: [libeufin] branch master updated: Dropping the integration tests.
Date: Thu, 03 Jun 2021 16:51:56 +0200

This is an automated email from the git hooks/post-receive script.

ms pushed a commit to branch master
in repository libeufin.

The following commit(s) were added to refs/heads/master by this push:
     new 48d6d1c  Dropping the integration tests.
48d6d1c is described below

commit 48d6d1c6acc572387159780ed585d63b3dd5da69
Author: MS <ms@taler.net>
AuthorDate: Thu Jun 3 16:51:32 2021 +0200

    Dropping the integration tests.
    
    They happen into the TypeScript harness at wallet.git now.
---
 Makefile                         |   5 -
 integration-tests/json_checks.py | 131 ---------
 integration-tests/tests.py       | 554 ---------------------------------------
 integration-tests/util.py        | 211 ---------------
 4 files changed, 901 deletions(-)

diff --git a/Makefile b/Makefile
index 7b00e5c..e01ef6f 100644
--- a/Makefile
+++ b/Makefile
@@ -37,11 +37,6 @@ check:
        @./gradlew check
 
 
-.PHONY: tests
-tests:
-       @cd integration-tests; py.test -k "not test_env" tests.py
-
-
 .PHONY: parse
 parse:
        @cd parsing-tests; py.test -s checks.py
diff --git a/integration-tests/json_checks.py b/integration-tests/json_checks.py
deleted file mode 100644
index 916712a..0000000
--- 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 ab7720b..0000000
--- 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 4960611..0000000
--- 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

-- 
To stop receiving notification emails like this one, please contact
gnunet@gnunet.org.



reply via email to

[Prev in Thread] Current Thread [Next in Thread]