diff --git a/apps/cli/pisa_cli.py b/apps/cli/pisa_cli.py index 7e0fc88..8692e22 100644 --- a/apps/cli/pisa_cli.py +++ b/apps/cli/pisa_cli.py @@ -28,6 +28,8 @@ from apps.cli import ( logger, ) +from common.constants import LOCATOR_LEN_HEX + HTTP_OK = 200 @@ -91,7 +93,7 @@ def load_private_key(sk_pem): def compute_locator(tx_id): - return tx_id[:32] + return tx_id[:LOCATOR_LEN_HEX] # returning True or False accordingly. diff --git a/pisa/api.py b/pisa/api.py index 32dd51a..ae9070f 100644 --- a/pisa/api.py +++ b/pisa/api.py @@ -8,7 +8,8 @@ from pisa.logger import Logger from pisa.inspector import Inspector from pisa.appointment import Appointment from pisa.block_processor import BlockProcessor -from common.constants import HTTP_OK, HTTP_BAD_REQUEST, HTTP_SERVICE_UNAVAILABLE + +from common.constants import HTTP_OK, HTTP_BAD_REQUEST, HTTP_SERVICE_UNAVAILABLE, LOCATOR_LEN_HEX # ToDo: #5-add-async-to-api @@ -74,7 +75,7 @@ def get_appointment(): response = [] # ToDo: #15-add-system-monitor - if not isinstance(locator, str) or len(locator) != 32: + if not isinstance(locator, str) or len(locator) != LOCATOR_LEN_HEX: response.append({"locator": locator, "status": "not_found"}) return jsonify(response) diff --git a/test/unit/test_api.py b/test/unit/test_api.py index 73d390a..0150af4 100644 --- a/test/unit/test_api.py +++ b/test/unit/test_api.py @@ -9,8 +9,11 @@ from pisa.watcher import Watcher from pisa.tools import bitcoin_cli from pisa import HOST, PORT, c_logger from pisa.conf import MAX_APPOINTMENTS + from test.unit.conftest import generate_block, generate_blocks, get_random_value_hex, generate_dummy_appointment_data +from common.constants import LOCATOR_LEN_BYTES + c_logger.disabled = True PISA_API = "http://{}:{}".format(HOST, PORT) @@ -61,7 +64,7 @@ def test_add_appointment(run_api, run_bitcoind, new_appt_data): def test_request_random_appointment(): - r = requests.get(url=PISA_API + "/get_appointment?locator=" + get_random_value_hex(32)) + r = requests.get(url=PISA_API + "/get_appointment?locator=" + get_random_value_hex(LOCATOR_LEN_BYTES)) assert r.status_code == 200 received_appointments = json.loads(r.content) diff --git a/test/unit/test_appointment.py b/test/unit/test_appointment.py index c50badb..d4e32bb 100644 --- a/test/unit/test_appointment.py +++ b/test/unit/test_appointment.py @@ -4,15 +4,19 @@ from pytest import fixture from pisa import c_logger from pisa.appointment import Appointment from pisa.encrypted_blob import EncryptedBlob + from test.unit.conftest import get_random_value_hex +from common.constants import LOCATOR_LEN_BYTES + c_logger.disabled = True + # Not much to test here, adding it for completeness @fixture def appointment_data(): - locator = get_random_value_hex(32) + locator = get_random_value_hex(LOCATOR_LEN_BYTES) start_time = 100 end_time = 120 dispute_delta = 20 diff --git a/test/unit/test_db_manager.py b/test/unit/test_db_manager.py index b1ae2bb..64824e9 100644 --- a/test/unit/test_db_manager.py +++ b/test/unit/test_db_manager.py @@ -5,9 +5,12 @@ import shutil from uuid import uuid4 from pisa.db_manager import DBManager -from test.unit.conftest import get_random_value_hex, generate_dummy_appointment from pisa.db_manager import WATCHER_LAST_BLOCK_KEY, RESPONDER_LAST_BLOCK_KEY, LOCATOR_MAP_PREFIX +from common.constants import LOCATOR_LEN_BYTES + +from test.unit.conftest import get_random_value_hex, generate_dummy_appointment + @pytest.fixture(scope="module") def watcher_appointments(): @@ -16,7 +19,7 @@ def watcher_appointments(): @pytest.fixture(scope="module") def responder_jobs(): - return {get_random_value_hex(32): get_random_value_hex(32) for _ in range(10)} + return {get_random_value_hex(16): get_random_value_hex(32) for _ in range(10)} def open_create_db(db_path): @@ -64,7 +67,7 @@ def test_load_appointments_db(db_manager): # We can add a bunch of data to the db and try again (data is stored in json by the manager) local_appointments = {} for _ in range(10): - key = get_random_value_hex(32) + key = get_random_value_hex(16) value = get_random_value_hex(32) local_appointments[key] = value @@ -105,7 +108,7 @@ def test_get_last_known_block(): def test_create_entry(db_manager): - key = get_random_value_hex(32) + key = get_random_value_hex(16) value = get_random_value_hex(32) # Adding a value with no prefix (create entry encodes values in utf-8 internally) @@ -115,7 +118,7 @@ def test_create_entry(db_manager): assert db_manager.db.get(key.encode("utf-8")).decode("utf-8") == value # If we prefix the key we should be able to get it if we add the prefix, but not otherwise - key = get_random_value_hex(32) + key = get_random_value_hex(16) prefix = "w" db_manager.create_entry(key, value, prefix=prefix) @@ -139,7 +142,7 @@ def test_delete_entry(db_manager): # Let's check that the same works if a prefix is provided. prefix = "r" - key = get_random_value_hex(32) + key = get_random_value_hex(16) value = get_random_value_hex(32) db_manager.create_entry(key, value, prefix) @@ -160,12 +163,12 @@ def test_load_responder_jobs_empty(db_manager): def test_load_locator_map_empty(db_manager): - assert db_manager.load_locator_map(get_random_value_hex(32)) is None + assert db_manager.load_locator_map(get_random_value_hex(LOCATOR_LEN_BYTES)) is None def test_store_update_locator_map_empty(db_manager): uuid = uuid4().hex - locator = get_random_value_hex(32) + locator = get_random_value_hex(LOCATOR_LEN_BYTES) db_manager.store_update_locator_map(locator, uuid) # Check that the locator map has been properly stored diff --git a/test/unit/test_inspector.py b/test/unit/test_inspector.py index 500c146..b191ddb 100644 --- a/test/unit/test_inspector.py +++ b/test/unit/test_inspector.py @@ -6,23 +6,42 @@ from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec from apps.cli.pisa_cli import build_appointment + from pisa import c_logger from pisa.errors import * from pisa.inspector import Inspector from pisa.appointment import Appointment from pisa.block_processor import BlockProcessor +from pisa.conf import MIN_DISPUTE_DELTA + from test.unit.conftest import get_random_value_hex -from pisa.conf import MIN_DISPUTE_DELTA +from common.constants import LOCATOR_LEN_BYTES, LOCATOR_LEN_HEX c_logger.disabled = True inspector = Inspector() APPOINTMENT_OK = (0, None) -NO_HEX_STRINGS = ["R" * 32, get_random_value_hex(15) + "PP", "$" * 32, " " * 32] -WRONG_TYPES = [[], "", get_random_value_hex(16), 3.2, 2.0, (), object, {}, " " * 32, object()] -WRONG_TYPES_NO_STR = [[], unhexlify(get_random_value_hex(16)), 3.2, 2.0, (), object, {}, object()] +NO_HEX_STRINGS = [ + "R" * LOCATOR_LEN_HEX, + get_random_value_hex(LOCATOR_LEN_BYTES - 1) + "PP", + "$" * LOCATOR_LEN_HEX, + " " * LOCATOR_LEN_HEX, +] +WRONG_TYPES = [ + [], + "", + get_random_value_hex(LOCATOR_LEN_BYTES), + 3.2, + 2.0, + (), + object, + {}, + " " * LOCATOR_LEN_HEX, + object(), +] +WRONG_TYPES_NO_STR = [[], unhexlify(get_random_value_hex(LOCATOR_LEN_BYTES)), 3.2, 2.0, (), object, {}, object()] def sign_appointment(sk, appointment): @@ -32,15 +51,15 @@ def sign_appointment(sk, appointment): def test_check_locator(): # Right appointment type, size and format - locator = get_random_value_hex(16) + locator = get_random_value_hex(LOCATOR_LEN_BYTES) assert Inspector.check_locator(locator) == APPOINTMENT_OK # Wrong size (too big) - locator = get_random_value_hex(17) + locator = get_random_value_hex(LOCATOR_LEN_BYTES + 1) assert Inspector.check_locator(locator)[0] == APPOINTMENT_WRONG_FIELD_SIZE # Wrong size (too small) - locator = get_random_value_hex(15) + locator = get_random_value_hex(LOCATOR_LEN_BYTES - 1) assert Inspector.check_locator(locator)[0] == APPOINTMENT_WRONG_FIELD_SIZE # Empty @@ -196,7 +215,7 @@ def test_inspect(run_bitcoind, generate_keypair): assert type(appointment) == tuple and appointment[0] != 0 # Valid appointment - locator = get_random_value_hex(16) + locator = get_random_value_hex(LOCATOR_LEN_BYTES) start_time = BlockProcessor.get_block_count() + 5 end_time = start_time + 20 dispute_delta = MIN_DISPUTE_DELTA