aljaz/python-teos
1
0
Fork 0
mirror of https://github.com/aljazceru/python-teos.git synced 2025-12-18 06:34:19 +01:00
Code Issues Packages Projects Releases Wiki Activity

Refactors the codebase to match the new naming from 793d563b8c053136dd5f936b0ef0cd88af215b06 and 3046eae38ee3f34857c96e6b9d43e645f7d2de2d

Browse Source
This commit is contained in:
Sergi Delgado Segura
2019-12-12 17:18:44 +01:00
parent 7c1986dfc8
commit d881706eb3
12 changed files with 144 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
test/unit/conftest.py
View File

@@ -11,7 +11,7 @@ from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization
from apps.cli.blob import Blob
from pisa.responder import Job
from pisa.responder import TransactionTracker
from pisa.watcher import Watcher
from pisa.tools import bitcoin_cli
from pisa.db_manager import DBManager
@@ -126,13 +126,13 @@ def generate_dummy_appointment(real_height=True, start_time_offset=5, end_time_o
return Appointment.from_dict(appointment_data["appointment"]), dispute_tx
def generate_dummy_job():
def generate_dummy_tracker():
dispute_txid = get_random_value_hex(32)
penalty_txid = get_random_value_hex(32)
penalty_rawtx = get_random_value_hex(100)
locator = dispute_txid[:LOCATOR_LEN_HEX]
job_data = dict(
tracker_data = dict(
locator=locator,
dispute_txid=dispute_txid,
penalty_txid=penalty_txid,
@@ -140,4 +140,4 @@ def generate_dummy_job():
appointment_end=100,
)
return Job.from_dict(job_data)
return TransactionTracker.from_dict(tracker_data)

6
test/unit/test_api.py
View File

@@ -109,7 +109,7 @@ def test_get_all_appointments_watcher():
local_locators = [appointment["locator"] for appointment in appointments]
assert set(watcher_locators) == set(local_locators)
assert len(received_appointments["responder_jobs"]) == 0
assert len(received_appointments["responder_trackers"]) == 0
def test_get_all_appointments_responder():
@@ -127,10 +127,10 @@ def test_get_all_appointments_responder():
received_appointments = json.loads(r.content)
# Make sure there is not pending locator in the watcher
responder_jobs = [v["locator"] for k, v in received_appointments["responder_jobs"].items()]
responder_trackers = [v["locator"] for k, v in received_appointments["responder_trackers"].items()]
local_locators = [appointment["locator"] for appointment in appointments]
assert set(responder_jobs) == set(local_locators)
assert set(responder_trackers) == set(local_locators)
assert len(received_appointments["watcher_appointments"]) == 0

38
test/unit/test_builder.py
View File

@@ -1,7 +1,7 @@
from uuid import uuid4
from pisa.builder import Builder
from test.unit.conftest import get_random_value_hex, generate_dummy_appointment, generate_dummy_job
from test.unit.conftest import get_random_value_hex, generate_dummy_appointment, generate_dummy_tracker
def test_build_appointments():
@@ -33,33 +33,33 @@ def test_build_appointments():
assert uuid in locator_uuid_map[appointment.locator]
def test_build_jobs():
jobs_data = {}
def test_build_trackers():
trackers_data = {}
# Create some jobs data
# Create some trackers data
for i in range(10):
job = generate_dummy_job()
tracker = generate_dummy_tracker()
jobs_data[uuid4().hex] = job.to_dict()
trackers_data[uuid4().hex] = tracker.to_dict()
# Add some additional jobs that share the same locator to test all the builder's cases
# Add some additional trackers that share the same locator to test all the builder's cases
if i % 2 == 0:
penalty_txid = job.penalty_txid
job = generate_dummy_job()
job.penalty_txid = penalty_txid
penalty_txid = tracker.penalty_txid
tracker = generate_dummy_tracker()
tracker.penalty_txid = penalty_txid
jobs_data[uuid4().hex] = job.to_dict()
trackers_data[uuid4().hex] = tracker.to_dict()
jobs, tx_job_map = Builder.build_jobs(jobs_data)
trackers, tx_tracker_map = Builder.build_trackers(trackers_data)
# Check that the built jobs match the data
for uuid, job in jobs.items():
assert uuid in jobs_data.keys()
job_dict = job.to_dict()
# Check that the built trackers match the data
for uuid, tracker in trackers.items():
assert uuid in trackers_data.keys()
tracker_dict = tracker.to_dict()
# The locator is not part of the job_data found in the database (for now)
assert jobs_data[uuid] == job_dict
assert uuid in tx_job_map[job.penalty_txid]
# The locator is not part of the tracker_data found in the database (for now)
assert trackers_data[uuid] == tracker_dict
assert uuid in tx_tracker_map[tracker.penalty_txid]
def test_build_block_queue():

74
test/unit/test_cleaner.py
View File

@@ -2,7 +2,7 @@ import random
from uuid import uuid4
from pisa import c_logger
from pisa.responder import Job
from pisa.responder import TransactionTracker
from pisa.cleaner import Cleaner
from pisa.appointment import Appointment
from pisa.db_manager import WATCHER_PREFIX
@@ -48,11 +48,11 @@ def set_up_appointments(db_manager, total_appointments):
return appointments, locator_uuid_map
def set_up_jobs(db_manager, total_jobs):
jobs = dict()
tx_job_map = dict()
def set_up_trackers(db_manager, total_trackers):
trackers = dict()
tx_tracker_map = dict()
for i in range(total_jobs):
for i in range(total_trackers):
uuid = uuid4().hex
# We use the same txid for penalty and dispute here, it shouldn't matter
@@ -61,24 +61,24 @@ def set_up_jobs(db_manager, total_jobs):
locator = dispute_txid[:LOCATOR_LEN_HEX]
# Assign both penalty_txid and dispute_txid the same id (it shouldn't matter)
job = Job(locator, dispute_txid, penalty_txid, None, None)
jobs[uuid] = job
tx_job_map[penalty_txid] = [uuid]
tracker = TransactionTracker(locator, dispute_txid, penalty_txid, None, None)
trackers[uuid] = tracker
tx_tracker_map[penalty_txid] = [uuid]
db_manager.store_responder_job(uuid, job.to_json())
db_manager.store_update_locator_map(job.locator, uuid)
db_manager.store_responder_tracker(uuid, tracker.to_json())
db_manager.store_update_locator_map(tracker.locator, uuid)
# Each penalty_txid can have more than one uuid assigned to it.
if i % 2:
uuid = uuid4().hex
jobs[uuid] = job
tx_job_map[penalty_txid].append(uuid)
trackers[uuid] = tracker
tx_tracker_map[penalty_txid].append(uuid)
db_manager.store_responder_job(uuid, job.to_json())
db_manager.store_update_locator_map(job.locator, uuid)
db_manager.store_responder_tracker(uuid, tracker.to_json())
db_manager.store_update_locator_map(tracker.locator, uuid)
return jobs, tx_job_map
return trackers, tx_tracker_map
def test_delete_expired_appointment(db_manager):
@@ -107,38 +107,38 @@ def test_delete_completed_appointments(db_manager):
assert db_appointments[uuid]["triggered"] is True
def test_delete_completed_jobs_db_match(db_manager):
def test_delete_completed_trackers_db_match(db_manager):
height = 0
for _ in range(ITERATIONS):
jobs, tx_job_map = set_up_jobs(db_manager, MAX_ITEMS)
selected_jobs = random.sample(list(jobs.keys()), k=ITEMS)
trackers, tx_tracker_map = set_up_trackers(db_manager, MAX_ITEMS)
selected_trackers = random.sample(list(trackers.keys()), k=ITEMS)
completed_jobs = [(job, 6) for job in selected_jobs]
completed_trackers = [(tracker, 6) for tracker in selected_trackers]
Cleaner.delete_completed_jobs(completed_jobs, height, jobs, tx_job_map, db_manager)
Cleaner.delete_completed_trackers(completed_trackers, height, trackers, tx_tracker_map, db_manager)
assert not set(completed_jobs).issubset(jobs.keys())
assert not set(completed_trackers).issubset(trackers.keys())
def test_delete_completed_jobs_no_db_match(db_manager):
def test_delete_completed_trackers_no_db_match(db_manager):
height = 0
for _ in range(ITERATIONS):
jobs, tx_job_map = set_up_jobs(db_manager, MAX_ITEMS)
selected_jobs = random.sample(list(jobs.keys()), k=ITEMS)
trackers, tx_tracker_map = set_up_trackers(db_manager, MAX_ITEMS)
selected_trackers = random.sample(list(trackers.keys()), k=ITEMS)
# Let's change some uuid's by creating new jobs that are not included in the db and share a penalty_txid with
# another job that is stored in the db.
for uuid in selected_jobs[: ITEMS // 2]:
penalty_txid = jobs[uuid].penalty_txid
# Let's change some uuid's by creating new trackers that are not included in the db and share a penalty_txid
# with another tracker that is stored in the db.
for uuid in selected_trackers[: ITEMS // 2]:
penalty_txid = trackers[uuid].penalty_txid
dispute_txid = get_random_value_hex(32)
locator = dispute_txid[:LOCATOR_LEN_HEX]
new_uuid = uuid4().hex
jobs[new_uuid] = Job(locator, dispute_txid, penalty_txid, None, None)
tx_job_map[penalty_txid].append(new_uuid)
selected_jobs.append(new_uuid)
trackers[new_uuid] = TransactionTracker(locator, dispute_txid, penalty_txid, None, None)
tx_tracker_map[penalty_txid].append(new_uuid)
selected_trackers.append(new_uuid)
# Let's add some random data
for i in range(ITEMS // 2):
@@ -147,12 +147,12 @@ def test_delete_completed_jobs_no_db_match(db_manager):
dispute_txid = get_random_value_hex(32)
locator = dispute_txid[:LOCATOR_LEN_HEX]
jobs[uuid] = Job(locator, dispute_txid, penalty_txid, None, None)
tx_job_map[penalty_txid] = [uuid]
selected_jobs.append(uuid)
trackers[uuid] = TransactionTracker(locator, dispute_txid, penalty_txid, None, None)
tx_tracker_map[penalty_txid] = [uuid]
selected_trackers.append(uuid)
completed_jobs = [(job, 6) for job in selected_jobs]
completed_trackers = [(tracker, 6) for tracker in selected_trackers]
# We should be able to delete the correct ones and not fail in the others
Cleaner.delete_completed_jobs(completed_jobs, height, jobs, tx_job_map, db_manager)
assert not set(completed_jobs).issubset(jobs.keys())
Cleaner.delete_completed_trackers(completed_trackers, height, trackers, tx_tracker_map, db_manager)
assert not set(completed_trackers).issubset(trackers.keys())

34
test/unit/test_db_manager.py
View File

@@ -18,7 +18,7 @@ def watcher_appointments():
@pytest.fixture(scope="module")
def responder_jobs():
def responder_trackers():
return {get_random_value_hex(16): get_random_value_hex(32) for _ in range(10)}
@@ -158,8 +158,8 @@ def test_load_watcher_appointments_empty(db_manager):
assert len(db_manager.load_watcher_appointments()) == 0
def test_load_responder_jobs_empty(db_manager):
assert len(db_manager.load_responder_jobs()) == 0
def test_load_responder_trackers_empty(db_manager):
assert len(db_manager.load_responder_trackers()) == 0
def test_load_locator_map_empty(db_manager):
@@ -228,16 +228,16 @@ def test_store_load_triggered_appointment(db_manager):
assert uuid in db_manager.load_watcher_appointments(include_triggered=True)
def test_store_load_responder_jobs(db_manager, responder_jobs):
for key, value in responder_jobs.items():
db_manager.store_responder_job(key, json.dumps({"value": value}))
def test_store_load_responder_trackers(db_manager, responder_trackers):
for key, value in responder_trackers.items():
db_manager.store_responder_tracker(key, json.dumps({"value": value}))
db_responder_jobs = db_manager.load_responder_jobs()
db_responder_trackers = db_manager.load_responder_trackers()
values = [job["value"] for job in db_responder_jobs.values()]
values = [tracker["value"] for tracker in db_responder_trackers.values()]
assert responder_jobs.keys() == db_responder_jobs.keys()
assert set(responder_jobs.values()) == set(values) and len(responder_jobs) == len(values)
assert responder_trackers.keys() == db_responder_trackers.keys()
assert set(responder_trackers.values()) == set(values) and len(responder_trackers) == len(values)
def test_delete_watcher_appointment(db_manager, watcher_appointments):
@@ -252,16 +252,16 @@ def test_delete_watcher_appointment(db_manager, watcher_appointments):
assert len(db_watcher_appointments) == 0
def test_delete_responder_job(db_manager, responder_jobs):
def test_delete_responder_tracker(db_manager, responder_trackers):
# Same for the responder
db_responder_jobs = db_manager.load_responder_jobs()
assert len(db_responder_jobs) != 0
db_responder_trackers = db_manager.load_responder_trackers()
assert len(db_responder_trackers) != 0
for key in responder_jobs.keys():
db_manager.delete_responder_job(key)
for key in responder_trackers.keys():
db_manager.delete_responder_tracker(key)
db_responder_jobs = db_manager.load_responder_jobs()
assert len(db_responder_jobs) == 0
db_responder_trackers = db_manager.load_responder_trackers()
assert len(db_responder_trackers) == 0
def test_store_load_last_block_hash_watcher(db_manager):