mirror of
https://github.com/aljazceru/python-teos.git
synced 2025-12-17 22:24:23 +01:00
Move config options used by watcher and responder to the constructor
This commit is contained in:
@@ -7,7 +7,16 @@ from common.logger import Logger
|
|||||||
from pisa.api import API
|
from pisa.api import API
|
||||||
from pisa.watcher import Watcher
|
from pisa.watcher import Watcher
|
||||||
from pisa.builder import Builder
|
from pisa.builder import Builder
|
||||||
from pisa.conf import BTC_NETWORK, PISA_SECRET_KEY
|
from pisa.conf import (
|
||||||
|
BTC_NETWORK,
|
||||||
|
FEED_PROTOCOL,
|
||||||
|
FEED_ADDR,
|
||||||
|
FEED_PORT,
|
||||||
|
MAX_APPOINTMENTS,
|
||||||
|
EXPIRY_DELTA,
|
||||||
|
MIN_TO_SELF_DELAY,
|
||||||
|
PISA_SECRET_KEY,
|
||||||
|
)
|
||||||
from pisa.responder import Responder
|
from pisa.responder import Responder
|
||||||
from pisa.db_manager import DBManager
|
from pisa.db_manager import DBManager
|
||||||
from pisa.block_processor import BlockProcessor
|
from pisa.block_processor import BlockProcessor
|
||||||
@@ -52,7 +61,9 @@ if __name__ == "__main__":
|
|||||||
with open(PISA_SECRET_KEY, "rb") as key_file:
|
with open(PISA_SECRET_KEY, "rb") as key_file:
|
||||||
secret_key_der = key_file.read()
|
secret_key_der = key_file.read()
|
||||||
|
|
||||||
watcher = Watcher(db_manager, secret_key_der)
|
pisa_config = load_config(conf)
|
||||||
|
|
||||||
|
watcher = Watcher(db_manager, secret_key_der, config=pisa_config)
|
||||||
|
|
||||||
if len(watcher_appointments_data) == 0 and len(responder_trackers_data) == 0:
|
if len(watcher_appointments_data) == 0 and len(responder_trackers_data) == 0:
|
||||||
logger.info("Fresh bootstrap")
|
logger.info("Fresh bootstrap")
|
||||||
@@ -65,7 +76,7 @@ if __name__ == "__main__":
|
|||||||
last_block_responder = db_manager.load_last_block_hash_responder()
|
last_block_responder = db_manager.load_last_block_hash_responder()
|
||||||
|
|
||||||
# FIXME: 32-reorgs-offline dropped txs are not used at this point.
|
# FIXME: 32-reorgs-offline dropped txs are not used at this point.
|
||||||
responder = Responder(db_manager)
|
responder = Responder(db_manager, pisa_config)
|
||||||
last_common_ancestor_responder = None
|
last_common_ancestor_responder = None
|
||||||
missed_blocks_responder = None
|
missed_blocks_responder = None
|
||||||
|
|
||||||
|
|||||||
@@ -135,13 +135,14 @@ class Responder:
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, db_manager):
|
def __init__(self, db_manager, config):
|
||||||
self.trackers = dict()
|
self.trackers = dict()
|
||||||
self.tx_tracker_map = dict()
|
self.tx_tracker_map = dict()
|
||||||
self.unconfirmed_txs = []
|
self.unconfirmed_txs = []
|
||||||
self.missed_confirmations = dict()
|
self.missed_confirmations = dict()
|
||||||
self.asleep = True
|
self.asleep = True
|
||||||
self.block_queue = Queue()
|
self.block_queue = Queue()
|
||||||
|
self.config = config
|
||||||
self.zmq_subscriber = None
|
self.zmq_subscriber = None
|
||||||
self.db_manager = db_manager
|
self.db_manager = db_manager
|
||||||
|
|
||||||
@@ -271,7 +272,7 @@ class Responder:
|
|||||||
from ``bitcoind``. Block ids are received trough the ``block_queue``.
|
from ``bitcoind``. Block ids are received trough the ``block_queue``.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
self.zmq_subscriber = ZMQSubscriber(parent="Responder")
|
self.zmq_subscriber = ZMQSubscriber(self.config, parent="Responder")
|
||||||
self.zmq_subscriber.handle(self.block_queue)
|
self.zmq_subscriber.handle(self.block_queue)
|
||||||
|
|
||||||
def do_watch(self):
|
def do_watch(self):
|
||||||
|
|||||||
@@ -8,12 +8,14 @@ from pisa.conf import FEED_PROTOCOL, FEED_ADDR, FEED_PORT
|
|||||||
class ZMQSubscriber:
|
class ZMQSubscriber:
|
||||||
""" Adapted from https://github.com/bitcoin/bitcoin/blob/master/contrib/zmq/zmq_sub.py"""
|
""" Adapted from https://github.com/bitcoin/bitcoin/blob/master/contrib/zmq/zmq_sub.py"""
|
||||||
|
|
||||||
def __init__(self, parent):
|
def __init__(self, config, parent):
|
||||||
self.zmqContext = zmq.Context()
|
self.zmqContext = zmq.Context()
|
||||||
self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
|
self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
|
||||||
self.zmqSubSocket.setsockopt(zmq.RCVHWM, 0)
|
self.zmqSubSocket.setsockopt(zmq.RCVHWM, 0)
|
||||||
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashblock")
|
self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashblock")
|
||||||
self.zmqSubSocket.connect("%s://%s:%s" % (FEED_PROTOCOL, FEED_ADDR, FEED_PORT))
|
self.zmqSubSocket.connect(
|
||||||
|
"%s://%s:%s" % (config.get("FEED_PROTOCOL"), config.get("FEED_ADDR"), config.get("FEED_PORT"))
|
||||||
|
)
|
||||||
self.logger = Logger("ZMQSubscriber-{}".format(parent))
|
self.logger = Logger("ZMQSubscriber-{}".format(parent))
|
||||||
|
|
||||||
self.terminate = False
|
self.terminate = False
|
||||||
|
|||||||
@@ -10,7 +10,6 @@ from pisa.cleaner import Cleaner
|
|||||||
from pisa.responder import Responder
|
from pisa.responder import Responder
|
||||||
from pisa.block_processor import BlockProcessor
|
from pisa.block_processor import BlockProcessor
|
||||||
from pisa.utils.zmq_subscriber import ZMQSubscriber
|
from pisa.utils.zmq_subscriber import ZMQSubscriber
|
||||||
from pisa.conf import EXPIRY_DELTA, MAX_APPOINTMENTS
|
|
||||||
|
|
||||||
logger = Logger("Watcher")
|
logger = Logger("Watcher")
|
||||||
|
|
||||||
@@ -58,18 +57,18 @@ class Watcher:
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, db_manager, sk_der, responder=None, max_appointments=MAX_APPOINTMENTS):
|
def __init__(self, db_manager, sk_der, config, responder=None):
|
||||||
self.appointments = dict()
|
self.appointments = dict()
|
||||||
self.locator_uuid_map = dict()
|
self.locator_uuid_map = dict()
|
||||||
self.asleep = True
|
self.asleep = True
|
||||||
self.block_queue = Queue()
|
self.block_queue = Queue()
|
||||||
self.max_appointments = max_appointments
|
self.config = config
|
||||||
self.zmq_subscriber = None
|
self.zmq_subscriber = None
|
||||||
self.db_manager = db_manager
|
self.db_manager = db_manager
|
||||||
self.signing_key = Cryptographer.load_private_key_der(sk_der)
|
self.signing_key = Cryptographer.load_private_key_der(sk_der)
|
||||||
|
|
||||||
if not isinstance(responder, Responder):
|
if not isinstance(responder, Responder):
|
||||||
self.responder = Responder(db_manager)
|
self.responder = Responder(db_manager, self.config)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def compute_locator(tx_id):
|
def compute_locator(tx_id):
|
||||||
@@ -115,7 +114,7 @@ class Watcher:
|
|||||||
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
if len(self.appointments) < self.max_appointments:
|
if len(self.appointments) < self.config.get("MAX_APPOINTMENTS"):
|
||||||
uuid = uuid4().hex
|
uuid = uuid4().hex
|
||||||
self.appointments[uuid] = appointment
|
self.appointments[uuid] = appointment
|
||||||
|
|
||||||
@@ -157,7 +156,7 @@ class Watcher:
|
|||||||
trough the ``block_queue``.
|
trough the ``block_queue``.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
self.zmq_subscriber = ZMQSubscriber(parent="Watcher")
|
self.zmq_subscriber = ZMQSubscriber(self.config, parent="Watcher")
|
||||||
self.zmq_subscriber.handle(self.block_queue)
|
self.zmq_subscriber.handle(self.block_queue)
|
||||||
|
|
||||||
def do_watch(self):
|
def do_watch(self):
|
||||||
@@ -182,7 +181,7 @@ class Watcher:
|
|||||||
expired_appointments = [
|
expired_appointments = [
|
||||||
uuid
|
uuid
|
||||||
for uuid, appointment in self.appointments.items()
|
for uuid, appointment in self.appointments.items()
|
||||||
if block["height"] > appointment.end_time + EXPIRY_DELTA
|
if block["height"] > appointment.end_time + self.config.get("EXPIRY_DELTA")
|
||||||
]
|
]
|
||||||
|
|
||||||
Cleaner.delete_expired_appointment(
|
Cleaner.delete_expired_appointment(
|
||||||
|
|||||||
@@ -147,3 +147,26 @@ def generate_dummy_tracker():
|
|||||||
)
|
)
|
||||||
|
|
||||||
return TransactionTracker.from_dict(tracker_data)
|
return TransactionTracker.from_dict(tracker_data)
|
||||||
|
|
||||||
|
|
||||||
|
def get_config():
|
||||||
|
config = {
|
||||||
|
"BTC_RPC_USER": "username",
|
||||||
|
"BTC_RPC_PASSWD": "password",
|
||||||
|
"BTC_RPC_HOST": "localhost",
|
||||||
|
"BTC_RPC_PORT": 8332,
|
||||||
|
"BTC_NETWORK": "regtest",
|
||||||
|
"FEED_PROTOCOL": "tcp",
|
||||||
|
"FEED_ADDR": "127.0.0.1",
|
||||||
|
"FEED_PORT": 28332,
|
||||||
|
"MAX_APPOINTMENTS": 100,
|
||||||
|
"EXPIRY_DELTA": 6,
|
||||||
|
"MIN_TO_SELF_DELAY": 20,
|
||||||
|
"SERVER_LOG_FILE": "pisa.log",
|
||||||
|
"PISA_SECRET_KEY": "pisa_sk.der",
|
||||||
|
"CLIENT_LOG_FILE": "pisa.log",
|
||||||
|
"TEST_LOG_FILE": "test.log",
|
||||||
|
"DB_PATH": "appointments",
|
||||||
|
}
|
||||||
|
|
||||||
|
return config
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ from pisa.api import API
|
|||||||
from pisa.watcher import Watcher
|
from pisa.watcher import Watcher
|
||||||
from pisa.tools import bitcoin_cli
|
from pisa.tools import bitcoin_cli
|
||||||
from pisa import HOST, PORT
|
from pisa import HOST, PORT
|
||||||
from pisa.conf import MAX_APPOINTMENTS
|
from pisa.conf import MAX_APPOINTMENTS, EXPIRY_DELTA
|
||||||
|
|
||||||
from test.pisa.unit.conftest import (
|
from test.pisa.unit.conftest import (
|
||||||
generate_block,
|
generate_block,
|
||||||
@@ -17,6 +17,7 @@ from test.pisa.unit.conftest import (
|
|||||||
get_random_value_hex,
|
get_random_value_hex,
|
||||||
generate_dummy_appointment_data,
|
generate_dummy_appointment_data,
|
||||||
generate_keypair,
|
generate_keypair,
|
||||||
|
get_config,
|
||||||
)
|
)
|
||||||
|
|
||||||
from common.constants import LOCATOR_LEN_BYTES
|
from common.constants import LOCATOR_LEN_BYTES
|
||||||
@@ -37,7 +38,7 @@ def run_api(db_manager):
|
|||||||
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
||||||
encryption_algorithm=serialization.NoEncryption(),
|
encryption_algorithm=serialization.NoEncryption(),
|
||||||
)
|
)
|
||||||
watcher = Watcher(db_manager, sk_der)
|
watcher = Watcher(db_manager, sk_der, get_config())
|
||||||
|
|
||||||
api_thread = Thread(target=API(watcher).start)
|
api_thread = Thread(target=API(watcher).start)
|
||||||
api_thread.daemon = True
|
api_thread.daemon = True
|
||||||
|
|||||||
@@ -17,12 +17,12 @@ from common.tools import check_sha256_hex_format
|
|||||||
|
|
||||||
from bitcoind_mock.utils import sha256d
|
from bitcoind_mock.utils import sha256d
|
||||||
from bitcoind_mock.transaction import TX
|
from bitcoind_mock.transaction import TX
|
||||||
from test.pisa.unit.conftest import generate_block, generate_blocks, get_random_value_hex
|
from test.pisa.unit.conftest import generate_block, generate_blocks, get_random_value_hex, get_config
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="module")
|
@pytest.fixture(scope="module")
|
||||||
def responder(db_manager):
|
def responder(db_manager):
|
||||||
return Responder(db_manager)
|
return Responder(db_manager, get_config())
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture()
|
@pytest.fixture()
|
||||||
@@ -151,11 +151,12 @@ def test_init_responder(responder):
|
|||||||
assert type(responder.missed_confirmations) is dict and len(responder.missed_confirmations) == 0
|
assert type(responder.missed_confirmations) is dict and len(responder.missed_confirmations) == 0
|
||||||
assert responder.block_queue.empty()
|
assert responder.block_queue.empty()
|
||||||
assert responder.asleep is True
|
assert responder.asleep is True
|
||||||
|
assert type(responder.config) is dict
|
||||||
assert responder.zmq_subscriber is None
|
assert responder.zmq_subscriber is None
|
||||||
|
|
||||||
|
|
||||||
def test_handle_breach(db_manager):
|
def test_handle_breach(db_manager):
|
||||||
responder = Responder(db_manager)
|
responder = Responder(db_manager, get_config())
|
||||||
uuid = uuid4().hex
|
uuid = uuid4().hex
|
||||||
tracker = create_dummy_tracker()
|
tracker = create_dummy_tracker()
|
||||||
|
|
||||||
@@ -295,7 +296,7 @@ def test_do_subscribe(responder):
|
|||||||
|
|
||||||
|
|
||||||
def test_do_watch(temp_db_manager):
|
def test_do_watch(temp_db_manager):
|
||||||
responder = Responder(temp_db_manager)
|
responder = Responder(temp_db_manager, get_config())
|
||||||
responder.block_queue = Queue()
|
responder.block_queue = Queue()
|
||||||
|
|
||||||
zmq_thread = Thread(target=responder.do_subscribe)
|
zmq_thread = Thread(target=responder.do_subscribe)
|
||||||
@@ -351,7 +352,7 @@ def test_do_watch(temp_db_manager):
|
|||||||
|
|
||||||
|
|
||||||
def test_check_confirmations(temp_db_manager):
|
def test_check_confirmations(temp_db_manager):
|
||||||
responder = Responder(temp_db_manager)
|
responder = Responder(temp_db_manager, get_config())
|
||||||
responder.block_queue = Queue()
|
responder.block_queue = Queue()
|
||||||
|
|
||||||
zmq_thread = Thread(target=responder.do_subscribe)
|
zmq_thread = Thread(target=responder.do_subscribe)
|
||||||
@@ -414,7 +415,7 @@ def test_get_completed_trackers(db_manager):
|
|||||||
initial_height = bitcoin_cli().getblockcount()
|
initial_height = bitcoin_cli().getblockcount()
|
||||||
|
|
||||||
# Let's use a fresh responder for this to make it easier to compare the results
|
# Let's use a fresh responder for this to make it easier to compare the results
|
||||||
responder = Responder(db_manager)
|
responder = Responder(db_manager, get_config())
|
||||||
|
|
||||||
# A complete tracker is a tracker that has reached the appointment end with enough confirmations (> MIN_CONFIRMATIONS)
|
# A complete tracker is a tracker that has reached the appointment end with enough confirmations (> MIN_CONFIRMATIONS)
|
||||||
# We'll create three type of transactions: end reached + enough conf, end reached + no enough conf, end not reached
|
# We'll create three type of transactions: end reached + enough conf, end reached + no enough conf, end not reached
|
||||||
@@ -462,7 +463,7 @@ def test_get_completed_trackers(db_manager):
|
|||||||
|
|
||||||
|
|
||||||
def test_rebroadcast(db_manager):
|
def test_rebroadcast(db_manager):
|
||||||
responder = Responder(db_manager)
|
responder = Responder(db_manager, get_config())
|
||||||
responder.asleep = False
|
responder.asleep = False
|
||||||
|
|
||||||
txs_to_rebroadcast = []
|
txs_to_rebroadcast = []
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ from test.pisa.unit.conftest import (
|
|||||||
generate_dummy_appointment,
|
generate_dummy_appointment,
|
||||||
get_random_value_hex,
|
get_random_value_hex,
|
||||||
generate_keypair,
|
generate_keypair,
|
||||||
|
get_config,
|
||||||
)
|
)
|
||||||
from pisa.conf import EXPIRY_DELTA, MAX_APPOINTMENTS
|
from pisa.conf import EXPIRY_DELTA, MAX_APPOINTMENTS
|
||||||
|
|
||||||
@@ -36,7 +37,7 @@ sk_der = signing_key.private_bytes(
|
|||||||
|
|
||||||
@pytest.fixture(scope="module")
|
@pytest.fixture(scope="module")
|
||||||
def watcher(db_manager):
|
def watcher(db_manager):
|
||||||
return Watcher(db_manager, sk_der)
|
return Watcher(db_manager, sk_der, get_config())
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="module")
|
@pytest.fixture(scope="module")
|
||||||
@@ -72,7 +73,7 @@ def test_init(watcher):
|
|||||||
assert type(watcher.locator_uuid_map) is dict and len(watcher.locator_uuid_map) == 0
|
assert type(watcher.locator_uuid_map) is dict and len(watcher.locator_uuid_map) == 0
|
||||||
assert watcher.block_queue.empty()
|
assert watcher.block_queue.empty()
|
||||||
assert watcher.asleep is True
|
assert watcher.asleep is True
|
||||||
assert watcher.max_appointments == MAX_APPOINTMENTS
|
assert type(watcher.config) is dict
|
||||||
assert watcher.zmq_subscriber is None
|
assert watcher.zmq_subscriber is None
|
||||||
assert type(watcher.responder) is Responder
|
assert type(watcher.responder) is Responder
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user