summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMS <ms@taler.net>2021-06-03 16:51:32 +0200
committerMS <ms@taler.net>2021-06-03 16:51:32 +0200
commit48d6d1c6acc572387159780ed585d63b3dd5da69 (patch)
tree170af4cc440886cbaba4abc64c03cdaef518aca8
parentd527df40d0553bbe6c20bfac0754e8a0253d02d4 (diff)
downloadlibeufin-48d6d1c6acc572387159780ed585d63b3dd5da69.tar.gz
libeufin-48d6d1c6acc572387159780ed585d63b3dd5da69.tar.bz2
libeufin-48d6d1c6acc572387159780ed585d63b3dd5da69.zip
Dropping the integration tests.
They happen into the TypeScript harness at wallet.git now.
-rw-r--r--Makefile5
-rw-r--r--integration-tests/json_checks.py131
-rwxr-xr-xintegration-tests/tests.py554
-rw-r--r--integration-tests/util.py211
4 files changed, 0 insertions, 901 deletions
diff --git a/Makefile b/Makefile
index 7b00e5c9..e01ef6fb 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 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