mirror of
https://github.com/aljazceru/python-teos.git
synced 2026-02-01 12:44:25 +01:00
Adds flake8 and fixes style issues
This commit is contained in:
Sergi Delgado Segura
@@ -102,12 +102,12 @@ def test_setup_logging():
|
||||
f_log_suffix = "_file_log"
|
||||
c_log_suffix = "_console_log"
|
||||
|
||||
assert len(logging.getLogger(prefix + f_log_suffix).handlers) is 0
|
||||
assert len(logging.getLogger(prefix + c_log_suffix).handlers) is 0
|
||||
assert len(logging.getLogger(prefix + f_log_suffix).handlers) == 0
|
||||
assert len(logging.getLogger(prefix + c_log_suffix).handlers) == 0
|
||||
|
||||
setup_logging(log_file, prefix)
|
||||
|
||||
assert len(logging.getLogger(prefix + f_log_suffix).handlers) is 1
|
||||
assert len(logging.getLogger(prefix + c_log_suffix).handlers) is 1
|
||||
assert len(logging.getLogger(prefix + f_log_suffix).handlers) == 1
|
||||
assert len(logging.getLogger(prefix + c_log_suffix).handlers) == 1
|
||||
|
||||
os.remove(log_file)
|
||||
|
||||
@@ -1,302 +0,0 @@
|
||||
import json
|
||||
import binascii
|
||||
from time import sleep
|
||||
from riemann.tx import Tx
|
||||
|
||||
from cli import teos_cli, DATA_DIR, DEFAULT_CONF, CONF_FILE_NAME
|
||||
|
||||
import common.cryptographer
|
||||
from common.blob import Blob
|
||||
from common.logger import Logger
|
||||
from common.tools import compute_locator
|
||||
from common.appointment import Appointment
|
||||
from common.cryptographer import Cryptographer
|
||||
from teos.utils.auth_proxy import JSONRPCException
|
||||
from test.teos.e2e.conftest import (
|
||||
END_TIME_DELTA,
|
||||
build_appointment_data,
|
||||
get_random_value_hex,
|
||||
create_penalty_tx,
|
||||
run_teosd,
|
||||
get_config,
|
||||
)
|
||||
|
||||
cli_config = get_config(DATA_DIR, CONF_FILE_NAME, DEFAULT_CONF)
|
||||
common.cryptographer.logger = Logger(actor="Cryptographer", log_name_prefix="")
|
||||
|
||||
teos_base_endpoint = "http://{}:{}".format(cli_config.get("TEOS_SERVER"), cli_config.get("TEOS_PORT"))
|
||||
teos_add_appointment_endpoint = "{}/add_appointment".format(teos_base_endpoint)
|
||||
teos_get_appointment_endpoint = "{}/get_appointment".format(teos_base_endpoint)
|
||||
|
||||
# Run teosd
|
||||
teosd_process = run_teosd()
|
||||
|
||||
|
||||
def broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, addr):
|
||||
# Broadcast the commitment transaction and mine a block
|
||||
bitcoin_cli.sendrawtransaction(commitment_tx)
|
||||
bitcoin_cli.generatetoaddress(1, addr)
|
||||
|
||||
|
||||
def get_appointment_info(locator):
|
||||
# Check that the justice has been triggered (the appointment has moved from Watcher to Responder)
|
||||
sleep(1) # Let's add a bit of delay so the state can be updated
|
||||
return teos_cli.get_appointment(locator, teos_base_endpoint)
|
||||
|
||||
|
||||
def test_appointment_life_cycle(bitcoin_cli, create_txs):
|
||||
commitment_tx, penalty_tx = create_txs
|
||||
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 teos_cli.add_appointment([json.dumps(appointment_data)], teos_base_endpoint, cli_config) is True
|
||||
|
||||
appointment_info = get_appointment_info(locator)
|
||||
assert appointment_info is not None
|
||||
assert len(appointment_info) == 1
|
||||
assert appointment_info[0].get("status") == "being_watched"
|
||||
|
||||
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 len(appointment_info) == 1
|
||||
assert appointment_info[0].get("status") == "dispute_responded"
|
||||
|
||||
# It can be also checked by ensuring that the penalty transaction made it to the network
|
||||
penalty_tx_id = bitcoin_cli.decoderawtransaction(penalty_tx).get("txid")
|
||||
|
||||
try:
|
||||
bitcoin_cli.getrawtransaction(penalty_tx_id)
|
||||
assert True
|
||||
|
||||
except JSONRPCException:
|
||||
# If the transaction if not found.
|
||||
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)
|
||||
bitcoin_cli.generatetoaddress(1, new_addr)
|
||||
|
||||
appointment_info = get_appointment_info(locator)
|
||||
assert appointment_info[0].get("status") == "not_found"
|
||||
|
||||
|
||||
def test_appointment_malformed_penalty(bitcoin_cli, create_txs):
|
||||
# Lets start by creating two valid transaction
|
||||
commitment_tx, penalty_tx = create_txs
|
||||
|
||||
# Now we can modify the penalty so it is invalid when broadcast
|
||||
mod_penalty_tx = Tx.from_hex(penalty_tx)
|
||||
tx_in = mod_penalty_tx.tx_ins[0].copy(redeem_script=b"")
|
||||
mod_penalty_tx = mod_penalty_tx.copy(tx_ins=[tx_in])
|
||||
|
||||
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
|
||||
appointment_data = build_appointment_data(bitcoin_cli, commitment_tx_id, mod_penalty_tx.hex())
|
||||
locator = compute_locator(commitment_tx_id)
|
||||
|
||||
assert teos_cli.add_appointment([json.dumps(appointment_data)], teos_base_endpoint, cli_config) is True
|
||||
|
||||
# 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)
|
||||
appointment_info = get_appointment_info(locator)
|
||||
|
||||
assert appointment_info is not None
|
||||
assert len(appointment_info) == 1
|
||||
assert appointment_info[0].get("status") == "not_found"
|
||||
|
||||
|
||||
def test_appointment_wrong_key(bitcoin_cli, create_txs):
|
||||
# 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
|
||||
|
||||
# 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)
|
||||
|
||||
# We can't use teos_cli.add_appointment here since it computes the locator internally, so let's do it manually.
|
||||
# We will encrypt the blob using the random value and derive the locator from the commitment tx.
|
||||
appointment_data["locator"] = compute_locator(bitcoin_cli.decoderawtransaction(commitment_tx).get("txid"))
|
||||
appointment_data["encrypted_blob"] = Cryptographer.encrypt(Blob(penalty_tx), get_random_value_hex(32))
|
||||
appointment = Appointment.from_dict(appointment_data)
|
||||
|
||||
teos_pk, cli_sk, cli_pk_der = teos_cli.load_keys(
|
||||
cli_config.get("TEOS_PUBLIC_KEY"), cli_config.get("CLI_PRIVATE_KEY"), cli_config.get("CLI_PUBLIC_KEY")
|
||||
)
|
||||
hex_pk_der = binascii.hexlify(cli_pk_der)
|
||||
|
||||
signature = Cryptographer.sign(appointment.serialize(), cli_sk)
|
||||
data = {"appointment": appointment.to_dict(), "signature": signature, "public_key": hex_pk_der.decode("utf-8")}
|
||||
|
||||
# Send appointment to the server.
|
||||
response = teos_cli.post_appointment(data, teos_base_endpoint)
|
||||
response_json = teos_cli.process_post_appointment_response(response)
|
||||
|
||||
# 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
|
||||
|
||||
# Trigger the appointment
|
||||
new_addr = bitcoin_cli.getnewaddress()
|
||||
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
|
||||
|
||||
# The appointment should have been removed since the decryption failed.
|
||||
sleep(1)
|
||||
appointment_info = get_appointment_info(appointment.locator)
|
||||
|
||||
assert appointment_info is not None
|
||||
assert len(appointment_info) == 1
|
||||
assert appointment_info[0].get("status") == "not_found"
|
||||
|
||||
|
||||
def test_two_identical_appointments(bitcoin_cli, create_txs):
|
||||
# Tests sending two identical appointments to the tower.
|
||||
# At the moment there are no checks for identical appointments, so both will be accepted, decrypted and kept until
|
||||
# the end.
|
||||
# TODO: 34-exact-duplicate-appointment
|
||||
# This tests sending an appointment with two valid transaction with the same locator.
|
||||
commitment_tx, penalty_tx = create_txs
|
||||
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 teos_cli.add_appointment([json.dumps(appointment_data)], teos_base_endpoint, cli_config) is True
|
||||
assert teos_cli.add_appointment([json.dumps(appointment_data)], teos_base_endpoint, cli_config) is True
|
||||
|
||||
# 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 first appointment should have made it to the Responder, and the second one should have been dropped for
|
||||
# double-spending
|
||||
sleep(1)
|
||||
appointment_info = get_appointment_info(locator)
|
||||
|
||||
assert appointment_info is not None
|
||||
assert len(appointment_info) == 2
|
||||
|
||||
for info in appointment_info:
|
||||
assert info.get("status") == "dispute_responded"
|
||||
assert info.get("penalty_rawtx") == penalty_tx
|
||||
|
||||
|
||||
def test_two_appointment_same_locator_different_penalty(bitcoin_cli, create_txs):
|
||||
# This tests sending an appointment with two valid transaction with the same locator.
|
||||
commitment_tx, penalty_tx1 = create_txs
|
||||
commitment_tx_id = bitcoin_cli.decoderawtransaction(commitment_tx).get("txid")
|
||||
|
||||
# We need to create a second penalty spending from the same commitment
|
||||
decoded_commitment_tx = bitcoin_cli.decoderawtransaction(commitment_tx)
|
||||
new_addr = bitcoin_cli.getnewaddress()
|
||||
penalty_tx2 = create_penalty_tx(bitcoin_cli, decoded_commitment_tx, new_addr)
|
||||
|
||||
appointment1_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx1)
|
||||
appointment2_data = build_appointment_data(bitcoin_cli, commitment_tx_id, penalty_tx2)
|
||||
locator = compute_locator(commitment_tx_id)
|
||||
|
||||
assert teos_cli.add_appointment([json.dumps(appointment1_data)], teos_base_endpoint, cli_config) is True
|
||||
assert teos_cli.add_appointment([json.dumps(appointment2_data)], teos_base_endpoint, cli_config) is True
|
||||
|
||||
# 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 first appointment should have made it to the Responder, and the second one should have been dropped for
|
||||
# double-spending
|
||||
sleep(1)
|
||||
appointment_info = get_appointment_info(locator)
|
||||
|
||||
assert appointment_info is not None
|
||||
assert len(appointment_info) == 1
|
||||
assert appointment_info[0].get("status") == "dispute_responded"
|
||||
assert appointment_info[0].get("penalty_rawtx") == penalty_tx1
|
||||
|
||||
|
||||
def test_appointment_shutdown_teos_trigger_back_online(create_txs, bitcoin_cli):
|
||||
global teosd_process
|
||||
|
||||
teos_pid = teosd_process.pid
|
||||
|
||||
commitment_tx, penalty_tx = create_txs
|
||||
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 teos_cli.add_appointment([json.dumps(appointment_data)], teos_base_endpoint, cli_config) is True
|
||||
|
||||
# Restart teos
|
||||
teosd_process.terminate()
|
||||
teosd_process = run_teosd()
|
||||
|
||||
assert teos_pid != teosd_process.pid
|
||||
|
||||
# Check that the appointment is still in the Watcher
|
||||
appointment_info = get_appointment_info(locator)
|
||||
|
||||
assert appointment_info is not None
|
||||
assert len(appointment_info) == 1
|
||||
assert appointment_info[0].get("status") == "being_watched"
|
||||
|
||||
# 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 len(appointment_info) == 1
|
||||
assert appointment_info[0].get("status") == "dispute_responded"
|
||||
|
||||
|
||||
def test_appointment_shutdown_teos_trigger_while_offline(create_txs, bitcoin_cli):
|
||||
global teosd_process
|
||||
|
||||
teos_pid = teosd_process.pid
|
||||
|
||||
commitment_tx, penalty_tx = create_txs
|
||||
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 teos_cli.add_appointment([json.dumps(appointment_data)], teos_base_endpoint, cli_config) is True
|
||||
|
||||
# Check that the appointment is still in the Watcher
|
||||
appointment_info = get_appointment_info(locator)
|
||||
assert appointment_info is not None
|
||||
assert len(appointment_info) == 1
|
||||
assert appointment_info[0].get("status") == "being_watched"
|
||||
|
||||
# Shutdown and trigger
|
||||
teosd_process.terminate()
|
||||
new_addr = bitcoin_cli.getnewaddress()
|
||||
broadcast_transaction_and_mine_block(bitcoin_cli, commitment_tx, new_addr)
|
||||
|
||||
# Restart
|
||||
teosd_process = run_teosd()
|
||||
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 len(appointment_info) == 1
|
||||
assert appointment_info[0].get("status") == "dispute_responded"
|
||||
|
||||
teosd_process.terminate()
|
||||
@@ -9,16 +9,9 @@ from teos.watcher import Watcher
|
||||
from teos.inspector import Inspector
|
||||
from teos.db_manager import DBManager
|
||||
from teos.gatekeeper import Gatekeeper
|
||||
from teos.chain_monitor import ChainMonitor
|
||||
from teos.responder import Responder, TransactionTracker
|
||||
|
||||
from test.teos.unit.conftest import (
|
||||
get_random_value_hex,
|
||||
generate_dummy_appointment,
|
||||
generate_keypair,
|
||||
get_config,
|
||||
bitcoind_feed_params,
|
||||
)
|
||||
from test.teos.unit.conftest import get_random_value_hex, generate_dummy_appointment, generate_keypair, get_config
|
||||
|
||||
from common.cryptographer import Cryptographer, hash_160
|
||||
from common.constants import (
|
||||
|
||||
@@ -1,6 +1,4 @@
|
||||
import pytest
|
||||
|
||||
from test.teos.unit.conftest import get_random_value_hex, generate_block, generate_blocks, fork, bitcoind_connect_params
|
||||
from test.teos.unit.conftest import get_random_value_hex, generate_block, generate_blocks, fork
|
||||
|
||||
|
||||
hex_tx = (
|
||||
|
||||
@@ -5,7 +5,7 @@ from threading import Thread, Event, Condition
|
||||
|
||||
from teos.chain_monitor import ChainMonitor
|
||||
|
||||
from test.teos.unit.conftest import get_random_value_hex, generate_block, bitcoind_connect_params, bitcoind_feed_params
|
||||
from test.teos.unit.conftest import get_random_value_hex, generate_block, bitcoind_feed_params
|
||||
|
||||
|
||||
def test_init(run_bitcoind, block_processor):
|
||||
@@ -64,8 +64,8 @@ def test_update_state(block_processor):
|
||||
|
||||
def test_monitor_chain_polling(db_manager, block_processor):
|
||||
# Try polling with the Watcher
|
||||
wq = Queue()
|
||||
chain_monitor = ChainMonitor(Queue(), Queue(), block_processor, bitcoind_feed_params)
|
||||
watcher_queue = Queue()
|
||||
chain_monitor = ChainMonitor(watcher_queue, Queue(), block_processor, bitcoind_feed_params)
|
||||
chain_monitor.best_tip = block_processor.get_best_block_hash()
|
||||
chain_monitor.polling_delta = 0.1
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import pytest
|
||||
from binascii import unhexlify
|
||||
|
||||
from teos.errors import *
|
||||
import teos.errors as errors
|
||||
from teos import LOG_PREFIX
|
||||
from teos.block_processor import BlockProcessor
|
||||
from teos.inspector import Inspector, InspectionFailed
|
||||
@@ -11,7 +11,7 @@ from common.logger import Logger
|
||||
from common.appointment import Appointment
|
||||
from common.constants import LOCATOR_LEN_BYTES, LOCATOR_LEN_HEX
|
||||
|
||||
from test.teos.unit.conftest import get_random_value_hex, generate_keypair, bitcoind_connect_params, get_config
|
||||
from test.teos.unit.conftest import get_random_value_hex, bitcoind_connect_params, get_config
|
||||
|
||||
common.cryptographer.logger = Logger(actor="Cryptographer", log_name_prefix=LOG_PREFIX)
|
||||
|
||||
@@ -53,7 +53,7 @@ def test_check_locator():
|
||||
inspector.check_locator(locator)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_WRONG_FIELD_SIZE
|
||||
assert e.erno == errors.APPOINTMENT_WRONG_FIELD_SIZE
|
||||
raise e
|
||||
|
||||
# Wrong size (too small)
|
||||
@@ -63,7 +63,7 @@ def test_check_locator():
|
||||
inspector.check_locator(locator)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_WRONG_FIELD_SIZE
|
||||
assert e.erno == errors.APPOINTMENT_WRONG_FIELD_SIZE
|
||||
raise e
|
||||
|
||||
# Empty
|
||||
@@ -73,7 +73,7 @@ def test_check_locator():
|
||||
inspector.check_locator(locator)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_EMPTY_FIELD
|
||||
assert e.erno == errors.APPOINTMENT_EMPTY_FIELD
|
||||
raise e
|
||||
|
||||
# Wrong type (several types tested, it should do for anything that is not a string)
|
||||
@@ -85,7 +85,7 @@ def test_check_locator():
|
||||
inspector.check_locator(locator)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_WRONG_FIELD_TYPE
|
||||
assert e.erno == errors.APPOINTMENT_WRONG_FIELD_TYPE
|
||||
raise e
|
||||
|
||||
# Wrong format (no hex)
|
||||
@@ -96,7 +96,7 @@ def test_check_locator():
|
||||
inspector.check_locator(locator)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_WRONG_FIELD_FORMAT
|
||||
assert e.erno == errors.APPOINTMENT_WRONG_FIELD_FORMAT
|
||||
raise e
|
||||
|
||||
|
||||
@@ -116,7 +116,7 @@ def test_check_start_time():
|
||||
inspector.check_start_time(start_time, current_time)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_FIELD_TOO_SMALL
|
||||
assert e.erno == errors.APPOINTMENT_FIELD_TOO_SMALL
|
||||
raise e
|
||||
|
||||
# Empty field
|
||||
@@ -126,7 +126,7 @@ def test_check_start_time():
|
||||
inspector.check_start_time(start_time, current_time)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_EMPTY_FIELD
|
||||
assert e.erno == errors.APPOINTMENT_EMPTY_FIELD
|
||||
raise e
|
||||
|
||||
# Wrong data type
|
||||
@@ -137,7 +137,7 @@ def test_check_start_time():
|
||||
inspector.check_start_time(start_time, current_time)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_WRONG_FIELD_TYPE
|
||||
assert e.erno == errors.APPOINTMENT_WRONG_FIELD_TYPE
|
||||
raise e
|
||||
|
||||
|
||||
@@ -158,7 +158,7 @@ def test_check_end_time():
|
||||
inspector.check_end_time(end_time, start_time, current_time)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_FIELD_TOO_SMALL
|
||||
assert e.erno == errors.APPOINTMENT_FIELD_TOO_SMALL
|
||||
raise e
|
||||
|
||||
# End time too small (either same height as current block or in the past)
|
||||
@@ -170,7 +170,7 @@ def test_check_end_time():
|
||||
inspector.check_end_time(end_time, start_time, current_time)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_FIELD_TOO_SMALL
|
||||
assert e.erno == errors.APPOINTMENT_FIELD_TOO_SMALL
|
||||
raise e
|
||||
|
||||
# Empty field
|
||||
@@ -180,7 +180,7 @@ def test_check_end_time():
|
||||
inspector.check_end_time(end_time, start_time, current_time)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_EMPTY_FIELD
|
||||
assert e.erno == errors.APPOINTMENT_EMPTY_FIELD
|
||||
raise e
|
||||
|
||||
# Wrong data type
|
||||
@@ -191,7 +191,7 @@ def test_check_end_time():
|
||||
inspector.check_end_time(end_time, start_time, current_time)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_WRONG_FIELD_TYPE
|
||||
assert e.erno == errors.APPOINTMENT_WRONG_FIELD_TYPE
|
||||
raise e
|
||||
|
||||
|
||||
@@ -209,7 +209,7 @@ def test_check_to_self_delay():
|
||||
inspector.check_to_self_delay(to_self_delay)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_FIELD_TOO_SMALL
|
||||
assert e.erno == errors.APPOINTMENT_FIELD_TOO_SMALL
|
||||
raise e
|
||||
|
||||
# Empty field
|
||||
@@ -219,7 +219,7 @@ def test_check_to_self_delay():
|
||||
inspector.check_to_self_delay(to_self_delay)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_EMPTY_FIELD
|
||||
assert e.erno == errors.APPOINTMENT_EMPTY_FIELD
|
||||
raise e
|
||||
|
||||
# Wrong data type
|
||||
@@ -230,7 +230,7 @@ def test_check_to_self_delay():
|
||||
inspector.check_to_self_delay(to_self_delay)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_WRONG_FIELD_TYPE
|
||||
assert e.erno == errors.APPOINTMENT_WRONG_FIELD_TYPE
|
||||
raise e
|
||||
|
||||
|
||||
@@ -251,7 +251,7 @@ def test_check_blob():
|
||||
inspector.check_blob(encrypted_blob)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_WRONG_FIELD_TYPE
|
||||
assert e.erno == errors.APPOINTMENT_WRONG_FIELD_TYPE
|
||||
raise e
|
||||
|
||||
# Empty field
|
||||
@@ -261,7 +261,7 @@ def test_check_blob():
|
||||
inspector.check_blob(encrypted_blob)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_EMPTY_FIELD
|
||||
assert e.erno == errors.APPOINTMENT_EMPTY_FIELD
|
||||
raise e
|
||||
|
||||
# Wrong format (no hex)
|
||||
@@ -272,7 +272,7 @@ def test_check_blob():
|
||||
inspector.check_blob(encrypted_blob)
|
||||
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_WRONG_FIELD_FORMAT
|
||||
assert e.erno == errors.APPOINTMENT_WRONG_FIELD_FORMAT
|
||||
raise e
|
||||
|
||||
|
||||
@@ -313,7 +313,7 @@ def test_inspect_wrong(run_bitcoind):
|
||||
inspector.inspect(data)
|
||||
except InspectionFailed as e:
|
||||
print(data)
|
||||
assert e.erno == APPOINTMENT_WRONG_FIELD
|
||||
assert e.erno == errors.APPOINTMENT_WRONG_FIELD
|
||||
raise e
|
||||
|
||||
# None data
|
||||
@@ -321,5 +321,5 @@ def test_inspect_wrong(run_bitcoind):
|
||||
try:
|
||||
inspector.inspect(None)
|
||||
except InspectionFailed as e:
|
||||
assert e.erno == APPOINTMENT_EMPTY_FIELD
|
||||
assert e.erno == errors.APPOINTMENT_EMPTY_FIELD
|
||||
raise e
|
||||
|
||||
Reference in New Issue
Block a user