aljaz/python-teos
1
0
Fork 0
mirror of https://github.com/aljazceru/python-teos.git synced 2026-02-05 14:44:22 +01:00
Code Issues Packages Projects Releases Wiki Activity

Refactors cli to avoid multi-type returns (normal return + None). Adds exceptions for errors.

Browse Source
This commit is contained in:
Sergi Delgado Segura
2020-04-03 21:57:38 +02:00
parent e2ae29b4fe
commit 4a65b2524b
5 changed files with 329 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
test/cli/unit/test_teos_cli.py
View File

@@ -1,19 +1,22 @@
import os
import json
import shutil
import pytest
import responses
from binascii import hexlify
from coincurve import PrivateKey
from requests.exceptions import ConnectionError, Timeout
from common.blob import Blob
import common.cryptographer
from common.logger import Logger
from common.tools import compute_locator
from common.appointment import Appointment
from common.cryptographer import Cryptographer
from common.blob import Blob
import cli.teos_cli as teos_cli
from cli.exceptions import InvalidParameter, InvalidKey, TowerResponseError
from test.cli.unit.conftest import get_random_value_hex, get_config
common.cryptographer.logger = Logger(actor="Cryptographer", log_name_prefix=teos_cli.LOG_PREFIX)
@@ -85,9 +88,7 @@ def test_add_appointment():
"available_slots": 100,
}
responses.add(responses.POST, add_appointment_endpoint, json=response, status=200)
result = teos_cli.add_appointment(
dummy_appointment_data, dummy_cli_sk, dummy_teos_pk, teos_url, config.get("APPOINTMENTS_FOLDER_NAME")
)
result = teos_cli.add_appointment(dummy_appointment_data, dummy_cli_sk, dummy_teos_pk, teos_url)
assert len(responses.calls) == 1
assert responses.calls[0].request.url == add_appointment_endpoint
@@ -106,13 +107,9 @@ def test_add_appointment_with_invalid_signature(monkeypatch):
}
responses.add(responses.POST, add_appointment_endpoint, json=response, status=200)
result = teos_cli.add_appointment(
dummy_appointment_data, dummy_cli_sk, dummy_teos_pk, teos_url, config.get("APPOINTMENTS_FOLDER_NAME")
)
assert result is False
shutil.rmtree(config.get("APPOINTMENTS_FOLDER_NAME"))
with pytest.raises(TowerResponseError):
teos_cli.add_appointment(dummy_appointment_data, dummy_cli_sk, dummy_teos_pk, teos_url)
@responses.activate
@@ -139,7 +136,8 @@ def test_get_appointment_err():
# Test that get_appointment handles a connection error appropriately.
responses.add(responses.POST, get_appointment_endpoint, body=ConnectionError())
assert not teos_cli.get_appointment(locator, dummy_cli_sk, dummy_teos_pk, teos_url)
with pytest.raises(ConnectionError):
teos_cli.get_appointment(locator, dummy_cli_sk, dummy_teos_pk, teos_url)
def test_load_keys():
@@ -151,7 +149,7 @@ def test_load_keys():
f.write(dummy_cli_sk.to_der())
with open(public_key_file_path, "wb") as f:
f.write(dummy_cli_compressed_pk)
with open(empty_file_path, "wb") as f:
with open(empty_file_path, "wb"):
pass
# Now we can test the function passing the using this files (we'll use the same pk for both)
@@ -159,25 +157,32 @@ def test_load_keys():
assert isinstance(r, tuple)
assert len(r) == 3
# If any param does not match we should get None as result
assert teos_cli.load_keys(None, private_key_file_path, public_key_file_path) is None
assert teos_cli.load_keys(public_key_file_path, None, public_key_file_path) is None
assert teos_cli.load_keys(public_key_file_path, private_key_file_path, None) is None
# If any param does not match the expected, we should get an InvalidKey exception
with pytest.raises(InvalidKey):
teos_cli.load_keys(None, private_key_file_path, public_key_file_path)
with pytest.raises(InvalidKey):
teos_cli.load_keys(public_key_file_path, None, public_key_file_path)
with pytest.raises(InvalidKey):
teos_cli.load_keys(public_key_file_path, private_key_file_path, None)
# The same should happen if we pass a public key where a private should be, for instance
assert teos_cli.load_keys(private_key_file_path, public_key_file_path, private_key_file_path) is None
with pytest.raises(InvalidKey):
teos_cli.load_keys(private_key_file_path, public_key_file_path, private_key_file_path)
# Same if any of the files is empty
assert teos_cli.load_keys(empty_file_path, private_key_file_path, public_key_file_path) is None
assert teos_cli.load_keys(public_key_file_path, empty_file_path, public_key_file_path) is None
assert teos_cli.load_keys(public_key_file_path, private_key_file_path, empty_file_path) is None
with pytest.raises(InvalidKey):
teos_cli.load_keys(empty_file_path, private_key_file_path, public_key_file_path)
with pytest.raises(InvalidKey):
teos_cli.load_keys(public_key_file_path, empty_file_path, public_key_file_path)
with pytest.raises(InvalidKey):
teos_cli.load_keys(public_key_file_path, private_key_file_path, empty_file_path)
# Remove the tmp files
os.remove(private_key_file_path)
os.remove(public_key_file_path)
os.remove(empty_file_path)
# WIP: HERE
@responses.activate
def test_post_request():
response = {
@@ -208,24 +213,23 @@ def test_process_post_response():
# If we modify the response code for a rejection (lets say 404) we should get None
responses.replace(responses.POST, add_appointment_endpoint, json=response, status=404)
r = teos_cli.post_request(json.dumps(dummy_appointment_data), add_appointment_endpoint)
assert teos_cli.process_post_response(r) is None
with pytest.raises(TowerResponseError):
r = teos_cli.post_request(json.dumps(dummy_appointment_data), add_appointment_endpoint)
teos_cli.process_post_response(r)
# The same should happen if the response is not in json
# The same should happen if the response is not in json independently of the return type
responses.replace(responses.POST, add_appointment_endpoint, status=404)
r = teos_cli.post_request(json.dumps(dummy_appointment_data), add_appointment_endpoint)
assert teos_cli.process_post_response(r) is None
with pytest.raises(TowerResponseError):
r = teos_cli.post_request(json.dumps(dummy_appointment_data), add_appointment_endpoint)
teos_cli.process_post_response(r)
responses.replace(responses.POST, add_appointment_endpoint, status=200)
with pytest.raises(TowerResponseError):
r = teos_cli.post_request(json.dumps(dummy_appointment_data), add_appointment_endpoint)
teos_cli.process_post_response(r)
def test_parse_add_appointment_args():
# If no args are passed, function should fail.
appt_data = teos_cli.parse_add_appointment_args(None)
assert not appt_data
# If file doesn't exist, function should fail.
appt_data = teos_cli.parse_add_appointment_args(["-f", "nonexistent_file"])
assert not appt_data
# If file exists and has data in it, function should work.
with open("appt_test_file", "w") as f:
json.dump(dummy_appointment_data, f)
@@ -233,12 +237,22 @@ def test_parse_add_appointment_args():
appt_data = teos_cli.parse_add_appointment_args(["-f", "appt_test_file"])
assert appt_data
os.remove("appt_test_file")
# If appointment json is passed in, function should work.
appt_data = teos_cli.parse_add_appointment_args([json.dumps(dummy_appointment_data)])
assert appt_data
os.remove("appt_test_file")
def test_parse_add_appointment_args_wrong():
# If no args are passed, function should fail.
with pytest.raises(InvalidParameter):
teos_cli.parse_add_appointment_args(None)
# If file doesn't exist, function should fail.
with pytest.raises(FileNotFoundError):
teos_cli.parse_add_appointment_args(["-f", "nonexistent_file"])
def test_save_appointment_receipt(monkeypatch):
appointments_folder = "test_appointments_receipts"

40
test/teos/e2e/conftest.py
View File

@@ -41,48 +41,32 @@ def setup_node(bitcoin_cli):
bitcoin_cli.generatetoaddress(106, new_addr)
@pytest.fixture()
def create_txs(bitcoin_cli):
def create_txs(bitcoin_cli, n=1):
utxos = bitcoin_cli.listunspent()
if len(utxos) == 0:
raise ValueError("There're no UTXOs.")
utxo = utxos.pop(0)
while utxo.get("amount") < Decimal(2 / pow(10, 5)):
utxo = utxos.pop(0)
signed_commitment_tx = create_commitment_tx(bitcoin_cli, utxo)
decoded_commitment_tx = bitcoin_cli.decoderawtransaction(signed_commitment_tx)
signed_penalty_tx = create_penalty_tx(bitcoin_cli, decoded_commitment_tx)
return signed_commitment_tx, signed_penalty_tx
@pytest.fixture()
def create_five_txs(bitcoin_cli):
utxos = bitcoin_cli.listunspent()
if len(utxos) < n:
raise ValueError("There're no enough UTXOs.")
signed_commitment_txs = []
signed_penalty_txs = []
for i in range(5):
if len(utxos) == 0:
raise ValueError("There're no UTXOs.")
for _ in range(n):
utxo = utxos.pop(0)
while utxo.get("amount") < Decimal(2 / pow(10, 5)):
utxo = utxos.pop(0)
signed_commitment_tx = create_commitment_tx(bitcoin_cli, utxo)
signed_commitment_txs.append(signed_commitment_tx)
decoded_commitment_tx = bitcoin_cli.decoderawtransaction(signed_commitment_tx)
signed_penalty_txs.append(create_penalty_tx(bitcoin_cli, decoded_commitment_tx))
signed_penalty_tx = create_penalty_tx(bitcoin_cli, decoded_commitment_tx)
return signed_commitment_txs, signed_penalty_txs
signed_commitment_txs.append(signed_commitment_tx)
signed_penalty_txs.append(signed_penalty_tx)
if len(signed_penalty_txs) > 1:
return signed_commitment_txs, signed_penalty_txs
else:
return signed_commitment_txs[0], signed_penalty_txs[0]
def run_teosd():

171
test/teos/e2e/test_basic_e2e.py
View File

@@ -1,9 +1,11 @@
import json
import pytest
from time import sleep
from riemann.tx import Tx
from binascii import hexlify
from coincurve import PrivateKey
from cli.exceptions import TowerResponseError
from cli import teos_cli, DATA_DIR, DEFAULT_CONF, CONF_FILE_NAME
import common.cryptographer
@@ -20,6 +22,7 @@ from test.teos.e2e.conftest import (
create_penalty_tx,
run_teosd,
get_config,
create_txs,
)
cli_config = get_config(DATA_DIR, CONF_FILE_NAME, DEFAULT_CONF)
@@ -50,9 +53,7 @@ def get_appointment_info(locator, sk=cli_sk):
def add_appointment(appointment_data, sk=cli_sk):
return teos_cli.add_appointment(
appointment_data, sk, teos_pk, teos_base_endpoint, cli_config.get("APPOINTMENTS_FOLDER_NAME")
)
return teos_cli.add_appointment(appointment_data, sk, teos_pk, teos_base_endpoint)
def get_all_appointments():
@@ -60,65 +61,72 @@ def get_all_appointments():
return json.loads(r)
def test_commands_non_registered(bitcoin_cli, create_txs):
def test_commands_non_registered(bitcoin_cli):
# All commands should fail if the user is not registered
# Add appointment
commitment_tx, penalty_tx = create_txs
commitment_tx, penalty_tx = create_txs(bitcoin_cli)
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
assert add_appointment(appointment_data) is False
with pytest.raises(TowerResponseError):
assert add_appointment(appointment_data)
# Get appointment
assert get_appointment_info(appointment_data.get("locator")) is None
with pytest.raises(TowerResponseError):
assert get_appointment_info(appointment_data.get("locator"))
def test_commands_registered(bitcoin_cli, create_txs):
def test_commands_registered(bitcoin_cli):
# Test registering and trying again
teos_cli.register(compressed_cli_pk, teos_base_endpoint)
# Add appointment
commitment_tx, penalty_tx = create_txs
commitment_tx, penalty_tx = create_txs(bitcoin_cli)
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
assert add_appointment(appointment_data) is True
appointment, available_slots = add_appointment(appointment_data)
assert isinstance(appointment, Appointment) and isinstance(available_slots, str)
# Get appointment
r = get_appointment_info(appointment_data.get("locator"))
assert r.get("locator") == appointment_data.get("locator")
assert r.get("appointment").get("locator") == appointment_data.get("locator")
assert r.get("appointment").get("encrypted_blob") == appointment_data.get("encrypted_blob")
assert r.get("appointment").get("start_time") == appointment_data.get("start_time")
assert r.get("appointment").get("end_time") == appointment_data.get("end_time")
assert r.get("locator") == appointment.locator
assert r.get("appointment") == appointment.to_dict()
def test_appointment_life_cycle(bitcoin_cli, create_txs):
def test_appointment_life_cycle(bitcoin_cli):
# First of all we need to register
# FIXME: requires register command in the cli
commitment_tx, penalty_tx = create_txs
teos_cli.register(compressed_cli_pk, teos_base_endpoint)
# After that we can build an appointment and send it to the tower
commitment_tx, penalty_tx = create_txs(bitcoin_cli)
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
locator = compute_locator(commitment_tx_id)
appointment, available_slots = add_appointment(appointment_data)
assert add_appointment(appointment_data) is True
# Get the information from the tower to check that it matches
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "being_watched"
assert appointment_info.get("locator") == locator
assert appointment_info.get("appointment") == appointment.to_dict()
# Check also the get_all_appointment endpoint
all_appointments = get_all_appointments()
watching = all_appointments.get("watcher_appointments")
responding = all_appointments.get("responder_trackers")
assert len(watching) == 1 and len(responding) == 0
# Trigger a breach and check again
new_addr = bitcoin_cli.getnewaddress()
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "dispute_responded"
assert appointment_info.get("locator") == locator
all_appointments = get_all_appointments()
watching = all_appointments.get("watcher_appointments")
@@ -137,29 +145,29 @@ def test_appointment_life_cycle(bitcoin_cli, create_txs):
assert False
# Now let's mine some blocks so the appointment reaches its end.
# Since we are running all the nodes remotely data may take more time than normal, and some confirmations may be
# missed, so we generate more than enough confirmations and add some delays.
for _ in range(int(1.5 * END_TIME_DELTA)):
sleep(1)
for _ in range(END_TIME_DELTA):
bitcoin_cli.generatetoaddress(1, new_addr)
assert get_appointment_info(locator) is None
# The appointment is no longer in the tower
with pytest.raises(TowerResponseError):
get_appointment_info(locator)
def test_multiple_appointments_life_cycle(bitcoin_cli, create_five_txs):
# Tests that get_all_appointments returns all the appointments the tower is storing at various stages in the appointment lifecycle.
def test_multiple_appointments_life_cycle(bitcoin_cli):
# Tests that get_all_appointments returns all the appointments the tower is storing at various stages in the
# appointment lifecycle.
appointments = []
commitment_txs, penalty_txs = create_five_txs
commitment_txs, penalty_txs = create_txs(bitcoin_cli, n=5)
# Create five appointments.
for i in range(5):
for commitment_tx, penalty_tx in zip(commitment_txs, penalty_txs):
appointment = {}
appointment["commitment_tx"] = commitment_txs[i]
appointment["penalty_tx"] = penalty_txs[i]
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_txs[i]).get("txid")
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_txs[i])
appointment["commitment_tx"] = commitment_tx
appointment["penalty_tx"] = penalty_tx
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
appointment["appointment_data"] = appointment_data
locator = compute_locator(commitment_tx_id)
appointment["locator"] = locator
@@ -171,10 +179,12 @@ def test_multiple_appointments_life_cycle(bitcoin_cli, create_five_txs):
add_appointment(appt.get("appointment_data"))
# Two of these appointments are breached, and the watchtower responds to them.
breached_appointments = []
for i in range(2):
new_addr = bitcoin_cli.getnewaddress()
broadcast_transaction_and_mine_block(bitcoin_cli, appointments[i]["commitment_tx"], new_addr)
bitcoin_cli.generatetoaddress(3, new_addr)
bitcoin_cli.generatetoaddress(1, new_addr)
breached_appointments.append(appointments[i]["locator"])
sleep(1)
# Test that they all show up in get_all_appointments at the correct stages.
@@ -182,6 +192,8 @@ def test_multiple_appointments_life_cycle(bitcoin_cli, create_five_txs):
watching = all_appointments.get("watcher_appointments")
responding = all_appointments.get("responder_trackers")
assert len(watching) == 3 and len(responding) == 2
responder_locators = [appointment["locator"] for uuid, appointment in responding.items()]
assert set(responder_locators) == set(breached_appointments)
# Now let's mine some blocks so these appointments reach the end of their lifecycle.
# Since we are running all the nodes remotely data may take more time than normal, and some confirmations may be
@@ -192,9 +204,9 @@ def test_multiple_appointments_life_cycle(bitcoin_cli, create_five_txs):
bitcoin_cli.generatetoaddress(1, new_addr)
def test_appointment_malformed_penalty(bitcoin_cli, create_txs):
def test_appointment_malformed_penalty(bitcoin_cli):
# Lets start by creating two valid transaction
commitment_tx, penalty_tx = create_txs
commitment_tx, penalty_tx = create_txs(bitcoin_cli)
# Now we can modify the penalty so it is invalid when broadcast
mod_penalty_tx = Tx.from_hex(penalty_tx)
@@ -205,21 +217,27 @@ def test_appointment_malformed_penalty(bitcoin_cli, create_txs):
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, mod_penalty_tx.hex())
locator = compute_locator(commitment_tx_id)
assert add_appointment(appointment_data) is True
appointment, _ = add_appointment(appointment_data)
# Get the information from the tower to check that it matches
appointment_info = get_appointment_info(locator)
assert appointment_info.get("status") == "being_watched"
assert appointment_info.get("locator") == locator
assert appointment_info.get("appointment") == appointment.to_dict()
# Broadcast the commitment transaction and mine a block
new_addr = bitcoin_cli.getnewaddress()
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
# The appointment should have been removed since the penalty_tx was malformed.
sleep(1)
assert get_appointment_info(locator) is None
with pytest.raises(TowerResponseError):
get_appointment_info(locator)
def test_appointment_wrong_key(bitcoin_cli, create_txs):
def test_appointment_wrong_decryption_key(bitcoin_cli):
# This tests an appointment encrypted with a key that has not been derived from the same source as the locator.
# Therefore the tower won't be able to decrypt the blob once the appointment is triggered.
commitment_tx, penalty_tx = create_txs
commitment_tx, penalty_tx = create_txs(bitcoin_cli)
# The appointment data is built using a random 32-byte value.
appointment_data = build_appointment_data(bitcoin_cli, get_random_value_hex(32), penalty_tx)
@@ -239,7 +257,6 @@ def test_appointment_wrong_key(bitcoin_cli, create_txs):
# Check that the server has accepted the appointment
signature = response_json.get("signature")
assert signature is not None
rpk = Cryptographer.recover_pk(appointment.serialize(), signature)
assert Cryptographer.verify_rpk(teos_pk, rpk) is True
assert response_json.get("locator") == appointment.locator
@@ -249,45 +266,43 @@ def test_appointment_wrong_key(bitcoin_cli, create_txs):
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
# The appointment should have been removed since the decryption failed.
sleep(1)
assert get_appointment_info(appointment.locator) is None
with pytest.raises(TowerResponseError):
get_appointment_info(appointment.locator)
def test_two_identical_appointments(bitcoin_cli, create_txs):
def test_two_identical_appointments(bitcoin_cli):
# Tests sending two identical appointments to the tower.
# This tests sending an appointment with two valid transaction with the same locator.
# If they come from the same user, the last one will be kept
commitment_tx, penalty_tx = create_txs
# If they come from the same user, the last one will be kept.
commitment_tx, penalty_tx = create_txs(bitcoin_cli)
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
locator = compute_locator(commitment_tx_id)
# Send the appointment twice
assert add_appointment(appointment_data) is True
assert add_appointment(appointment_data) is True
add_appointment(appointment_data)
add_appointment(appointment_data)
# Broadcast the commitment transaction and mine a block
new_addr = bitcoin_cli.getnewaddress()
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
# The last appointment should have made it to the Responder
sleep(1)
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "dispute_responded"
assert appointment_info.get("appointment").get("penalty_rawtx") == penalty_tx
# FIXME: This test won't work since we're still passing appointment replicas to the Responder.
# Uncomment when #88 is addressed
# def test_two_identical_appointments_different_users(bitcoin_cli, create_txs):
# def test_two_identical_appointments_different_users(bitcoin_cli):
# # Tests sending two identical appointments from different users to the tower.
# # This tests sending an appointment with two valid transaction with the same locator.
# # If they come from different users, both will be kept, but one will be dropped fro double-spending when passing to
# # the responder
# commitment_tx, penalty_tx = create_txs
# commitment_tx, penalty_tx = create_txs(bitcoin_cli)
# commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
#
# appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
@@ -328,9 +343,9 @@ def test_two_identical_appointments(bitcoin_cli, create_txs):
# assert appointment_info.get("appointment").get("penalty_rawtx") == penalty_tx
def test_two_appointment_same_locator_different_penalty_different_users(bitcoin_cli, create_txs):
# This tests sending an appointment with two valid transaction with the same locator.
commitment_tx, penalty_tx1 = create_txs
def test_two_appointment_same_locator_different_penalty_different_users(bitcoin_cli):
# This tests sending an appointment with two valid transaction with the same locator fro different users
commitment_tx, penalty_tx1 = create_txs(bitcoin_cli)
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
# We need to create a second penalty spending from the same commitment
@@ -342,28 +357,24 @@ def test_two_appointment_same_locator_different_penalty_different_users(bitcoin_
appointment2_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx2)
locator = compute_locator(commitment_tx_id)
# tmp keys from a different user
# tmp keys for a different user
tmp_sk = PrivateKey()
tmp_compressed_pk = hexlify(tmp_sk.public_key.format(compressed=True)).decode("utf-8")
teos_cli.register(tmp_compressed_pk, teos_base_endpoint)
assert add_appointment(appointment1_data) is True
assert add_appointment(appointment2_data, sk=tmp_sk) is True
appointment, _ = add_appointment(appointment1_data)
appointment_2, _ = add_appointment(appointment2_data, sk=tmp_sk)
# Broadcast the commitment transaction and mine a block
new_addr = bitcoin_cli.getnewaddress()
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
# One of the transactions must have made it to the Responder while the other must have been dropped for
# double-spending
sleep(1)
appointment_info = get_appointment_info(locator)
appointment2_info = get_appointment_info(locator, sk=tmp_sk)
# One of the two request must be None, while the other must be valid
assert (appointment_info is None and appointment2_info is not None) or (
appointment2_info is None and appointment_info is not None
)
# double-spending. That means that one of the responses from the tower should fail
appointment_info = None
with pytest.raises(TowerResponseError):
appointment_info = get_appointment_info(locator)
appointment2_info = get_appointment_info(locator, sk=tmp_sk)
if appointment_info is None:
appointment_info = appointment2_info
@@ -371,19 +382,20 @@ def test_two_appointment_same_locator_different_penalty_different_users(bitcoin_
assert appointment_info.get("status") == "dispute_responded"
assert appointment_info.get("locator") == appointment1_data.get("locator")
assert appointment_info.get("appointment").get("penalty_tx") == appointment1_data.get("penalty_tx")
def test_appointment_shutdown_teos_trigger_back_online(create_txs, bitcoin_cli):
def test_appointment_shutdown_teos_trigger_back_online(bitcoin_cli):
global teosd_process
teos_pid = teosd_process.pid
commitment_tx, penalty_tx = create_txs
commitment_tx, penalty_tx = create_txs(bitcoin_cli)
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
locator = compute_locator(commitment_tx_id)
assert add_appointment(appointment_data) is True
appointment, _ = add_appointment(appointment_data)
# Restart teos
teosd_process.terminate()
@@ -392,40 +404,36 @@ def test_appointment_shutdown_teos_trigger_back_online(create_txs, bitcoin_cli):
assert teos_pid != teosd_process.pid
# Check that the appointment is still in the Watcher
sleep(1)
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "being_watched"
assert appointment_info.get("appointment") == appointment.to_dict()
# Trigger appointment after restart
new_addr = bitcoin_cli.getnewaddress()
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
# The appointment should have been moved to the Responder
sleep(1)
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "dispute_responded"
def test_appointment_shutdown_teos_trigger_while_offline(create_txs, bitcoin_cli):
def test_appointment_shutdown_teos_trigger_while_offline(bitcoin_cli):
global teosd_process
teos_pid = teosd_process.pid
commitment_tx, penalty_tx = create_txs
commitment_tx, penalty_tx = create_txs(bitcoin_cli)
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx)
locator = compute_locator(commitment_tx_id)
assert add_appointment(appointment_data) is True
appointment, _ = add_appointment(appointment_data)
# Check that the appointment is still in the Watcher
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "being_watched"
assert appointment_info.get("appointment") == appointment.to_dict()
# Shutdown and trigger
teosd_process.terminate()
@@ -437,10 +445,7 @@ def test_appointment_shutdown_teos_trigger_while_offline(create_txs, bitcoin_cli
assert teos_pid != teosd_process.pid
# The appointment should have been moved to the Responder
sleep(1)
appointment_info = get_appointment_info(locator)
assert appointment_info is not None
assert appointment_info.get("status") == "dispute_responded"
teosd_process.terminate()