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

Adds unit test for watcher.get_appointment_summary

Browse Source
This commit is contained in:
Sergi Delgado Segura
2020-03-30 16:19:04 +02:00
parent 495ea3800e
commit f55e37b0fc
2 changed files with 18 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
teos/watcher.py
View File

@@ -39,8 +39,8 @@ class Watcher:
expiry_delta (:obj:`int`): the additional time the ``Watcher`` will keep an expired appointment around. expiry_delta (:obj:`int`): the additional time the ``Watcher`` will keep an expired appointment around.
Attributes: Attributes:
appointments (:obj:`dict`): a dictionary containing a simplification of the appointments (:obj:`Appointment appointments (:obj:`dict`): a dictionary containing a summary of the appointments (:obj:`Appointment
<teos.appointment.Appointment>` instances) accepted by the tower (``locator`` and ``end_time``). <teos.appointment.Appointment>` instances) accepted by the tower (``locator``, ``end_time``, and ``size``).
It's populated trough ``add_appointment``. It's populated trough ``add_appointment``.
locator_uuid_map (:obj:`dict`): a ``locator:uuid`` map used to allow the :obj:`Watcher` to deal with several locator_uuid_map (:obj:`dict`): a ``locator:uuid`` map used to allow the :obj:`Watcher` to deal with several
appointments with the same ``locator``. appointments with the same ``locator``.
@@ -71,6 +71,10 @@ class Watcher:
self.signing_key = Cryptographer.load_private_key_der(sk_der) self.signing_key = Cryptographer.load_private_key_der(sk_der)
def awake(self): def awake(self):
"""
Starts a new thread to monitor the blockchain for channel breaches.
"""
watcher_thread = Thread(target=self.do_watch, daemon=True) watcher_thread = Thread(target=self.do_watch, daemon=True)
watcher_thread.start() watcher_thread.start()
@@ -88,7 +92,6 @@ class Watcher:
:obj:`dict` or :obj:`None`: a dictionary with the appointment summary, or None if the appointment is not :obj:`dict` or :obj:`None`: a dictionary with the appointment summary, or None if the appointment is not
found. found.
""" """
return self.appointments.get(uuid) return self.appointments.get(uuid)
def add_appointment(self, appointment, user_pk): def add_appointment(self, appointment, user_pk):

13
test/teos/unit/test_watcher.py
View File

@@ -112,6 +112,17 @@ def test_init(run_bitcoind, watcher):
assert isinstance(watcher.signing_key, PrivateKey) assert isinstance(watcher.signing_key, PrivateKey)
def test_get_appointment_summary(watcher):
# get_appointment_summary returns an appointment summary if found, else None.
random_uuid = get_random_value_hex(16)
appointment_summary = {"locator": get_random_value_hex(16), "end_time": 10, "size": 200}
watcher.appointments[random_uuid] = appointment_summary
assert watcher.get_appointment_summary(random_uuid) == appointment_summary
# Requesting a non-existing appointment
assert watcher.get_appointment_summary(get_random_value_hex(16)) is None
def test_add_appointment(watcher): def test_add_appointment(watcher):
# We should be able to add appointments up to the limit # We should be able to add appointments up to the limit
for _ in range(10): for _ in range(10):
@@ -186,7 +197,7 @@ def test_do_watch(watcher, temp_db_manager):
watcher.appointments = {} watcher.appointments = {}
for uuid, appointment in appointments.items(): for uuid, appointment in appointments.items():
watcher.appointments[uuid] = {"locator": appointment.locator, "end_time": appointment.end_time} watcher.appointments[uuid] = {"locator": appointment.locator, "end_time": appointment.end_time, "size": 200}
watcher.db_manager.store_watcher_appointment(uuid, appointment.to_json()) watcher.db_manager.store_watcher_appointment(uuid, appointment.to_json())
watcher.db_manager.create_append_locator_map(appointment.locator, uuid) watcher.db_manager.create_append_locator_map(appointment.locator, uuid)