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

Addresses requested changes

Browse Source
This commit is contained in:
Sergi Delgado Segura
2019-11-22 14:48:38 +00:00
parent 2183c57f53
commit d2b7216a96
5 changed files with 49 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
test/unit/test_responder.py
View File

@@ -2,11 +2,13 @@ import json
import pytest
import random
from uuid import uuid4
from shutil import rmtree
from copy import deepcopy
from threading import Thread
from queue import Queue, Empty
from pisa import c_logger
from pisa.db_manager import DBManager
from test.simulator.utils import sha256d
from pisa.responder import Responder, Job
from test.simulator.bitcoind_sim import TX
@@ -22,6 +24,16 @@ def responder(db_manager):
return Responder(db_manager)
@pytest.fixture()
def temp_db_manager():
db_name = get_random_value_hex(8)
db_manager = DBManager(db_name)
yield db_manager
db_manager.db.close()
rmtree(db_name)
def create_dummy_job_data(random_txid=False, justice_rawtx=None):
# The following transaction data corresponds to a valid transaction. For some test it may be interesting to have
# some valid data, but for others we may need multiple different justice_txids.
@@ -274,12 +286,13 @@ def test_do_subscribe(responder):
assert False
def test_do_watch(responder):
# Reinitializing responder (but keeping the subscriber)
responder.jobs = dict()
responder.tx_job_map = dict()
responder.unconfirmed_txs = []
responder.missed_confirmations = dict()
def test_do_watch(temp_db_manager):
responder = Responder(temp_db_manager)
responder.block_queue = Queue()
zmq_thread = Thread(target=responder.do_subscribe)
zmq_thread.daemon = True
zmq_thread.start()
jobs = [create_dummy_job(justice_rawtx=TX.create_dummy_transaction()) for _ in range(20)]
@@ -329,12 +342,13 @@ def test_do_watch(responder):
assert responder.asleep is True
def test_check_confirmations(responder):
# Reinitializing responder (but keeping the subscriber)
responder.jobs = dict()
responder.tx_job_map = dict()
responder.unconfirmed_txs = []
responder.missed_confirmations = dict()
def test_check_confirmations(temp_db_manager):
responder = Responder(temp_db_manager)
responder.block_queue = Queue()
zmq_thread = Thread(target=responder.do_subscribe)
zmq_thread.daemon = True
zmq_thread.start()
# check_confirmations checks, given a list of transaction for a block, what of the known justice transaction have
# been confirmed. To test this we need to create a list of transactions and the state of the responder