diff --git a/src/seedsigner/gui/screens/psbt_screens.py b/src/seedsigner/gui/screens/psbt_screens.py index 330742d5..11851292 100644 --- a/src/seedsigner/gui/screens/psbt_screens.py +++ b/src/seedsigner/gui/screens/psbt_screens.py @@ -406,7 +406,7 @@ def draw_line_segment(curves, i, j, color): ) prev_color = reset_color - while self.keep_running: + while not self.event.wait(timeout=0.02): # No need to CPU limit when running in its own thread? with self.renderer.lock: # Only generate one new pulse at a time; trailing "reset_color" pulse # erases the most recent pulse. @@ -441,10 +441,6 @@ def draw_line_segment(curves, i, j, color): self.renderer.show_image() - # No need to CPU limit when running in its own thread? - time.sleep(0.02) - - @dataclass class PSBTMathScreen(ButtonListScreen): diff --git a/src/seedsigner/gui/screens/scan_screens.py b/src/seedsigner/gui/screens/scan_screens.py index 13d3dcf7..40f78eca 100644 --- a/src/seedsigner/gui/screens/scan_screens.py +++ b/src/seedsigner/gui/screens/scan_screens.py @@ -85,16 +85,19 @@ def __init__(self, camera: Camera, decoder: DecodeQR, renderer: renderer.Rendere def run(self): - from timeit import default_timer as timer - instructions_font = Fonts.get_font(GUIConstants.BODY_FONT_NAME, GUIConstants.BUTTON_FONT_SIZE) start_time = time.time() num_frames = 0 show_framerate = False # enable for debugging / testing - while self.keep_running: + while not self.event.is_set(): frame = self.camera.read_video_stream(as_image=True) - if frame is not None: + if frame is None: + # give the camera a moment to get started + time.sleep(0.1) + continue + + else: num_frames += 1 cur_time = time.time() cur_fps = num_frames / (cur_time - start_time) @@ -167,10 +170,17 @@ def _run(self): self.threads[0].decoder_fps = decoder_fps if status in (DecodeQRStatus.COMPLETE, DecodeQRStatus.INVALID): - self.camera.stop_video_stream_mode() break if self.hw_inputs.check_for_low(HardwareButtonsConstants.KEY_RIGHT) or self.hw_inputs.check_for_low(HardwareButtonsConstants.KEY_LEFT): - self.camera.stop_video_stream_mode() break + self.camera.stop_video_stream_mode() + + # Stop the LivePreviewThread and... + self.threads[-1].stop() + + # ...WAIT for it to exit so that it doesn't compete to render one last preview + # frame while we start rendering the next screen. + self.threads[-1].join() + diff --git a/src/seedsigner/gui/screens/screen.py b/src/seedsigner/gui/screens/screen.py index b229857d..23b877c3 100644 --- a/src/seedsigner/gui/screens/screen.py +++ b/src/seedsigner/gui/screens/screen.py @@ -10,8 +10,9 @@ from seedsigner.gui.keyboard import Keyboard, TextEntryDisplay from seedsigner.gui.renderer import Renderer from seedsigner.hardware.buttons import HardwareButtonsConstants, HardwareButtons +from seedsigner.models.encode_qr import EncodeQR from seedsigner.models.settings import SettingsConstants -from seedsigner.models.threads import BaseThread, ThreadsafeCounter +from seedsigner.models.threads import BaseThread, ThreadsafeCounter, ThreadsafeVar # Must be huge numbers to avoid conflicting with the selected_button returned by the @@ -150,7 +151,7 @@ def run(self): screen_y=int((renderer.canvas_height - bounding_box[3])/2), ).render() - while self.keep_running: + while not self.event.is_set(): with renderer.lock: # Render leading arc renderer.draw.arc( @@ -657,11 +658,10 @@ def swap_selected_button(new_selected_button: int): @dataclass class QRDisplayScreen(BaseScreen): - qr_encoder: 'EncodeQR' = None + qr_encoder: EncodeQR = None class QRDisplayThread(BaseThread): - def __init__(self, qr_encoder: 'EncodeQR', qr_brightness: ThreadsafeCounter, renderer: Renderer, - tips_start_time: ThreadsafeCounter): + def __init__(self, qr_encoder: EncodeQR, qr_brightness: ThreadsafeVar[int], renderer: Renderer, tips_start_time: ThreadsafeCounter): super().__init__() self.qr_encoder = qr_encoder self.qr_brightness = qr_brightness @@ -748,22 +748,19 @@ def run(self): # Loop whether the QR is a single frame or animated; each loop might adjust # brightness setting. - while self.keep_running: + while not self.event.wait(timeout=(5/30.0)): # Target n held frames per second before rendering next QR image # convert the self.qr_brightness integer (31-255) into hex triplets - hex_color = (hex(self.qr_brightness.cur_count).split('x')[1]) * 3 + hex_color = (hex(self.qr_brightness.cur_value).split('x')[1]) * 3 image = self.qr_encoder.next_part_image(240, 240, border=2, background_color=hex_color) # Display the brightness tips toast duration = 10 ** 9 * 1.2 # 1.2 seconds - if show_brightness_tips and time.time_ns() - self.tips_start_time.cur_count < duration: + if show_brightness_tips and time.time_ns() - self.tips_start_time.cur_value < duration: self.add_brightness_tips(image) with self.renderer.lock: self.renderer.show_image(image) - # Target n held frames per second before rendering next QR image - time.sleep(5 / 30.0) - def __post_init__(self): from seedsigner.models.settings import Settings @@ -771,7 +768,7 @@ def __post_init__(self): # Shared coordination var so the display thread can detect success settings = Settings.get_instance() - self.qr_brightness = ThreadsafeCounter( + self.qr_brightness = ThreadsafeVar[int]( initial_value=settings.get_value(SettingsConstants.SETTING__QR_BRIGHTNESS)) self.tips_start_time = ThreadsafeCounter(initial_value=time.time_ns()) @@ -799,12 +796,12 @@ def _run(self): ) if user_input == HardwareButtonsConstants.KEY_DOWN: # Reduce QR code background brightness - self.qr_brightness.set_value(max(31, self.qr_brightness.cur_count - 31)) + self.qr_brightness.set_value(max(31, self.qr_brightness.cur_value - 31)) self.tips_start_time.set_value(time.time_ns()) elif user_input == HardwareButtonsConstants.KEY_UP: # Incrase QR code background brightness - self.qr_brightness.set_value(min(self.qr_brightness.cur_count + 31, 255)) + self.qr_brightness.set_value(min(self.qr_brightness.cur_value + 31, 255)) self.tips_start_time.set_value(time.time_ns()) else: @@ -814,7 +811,7 @@ def _run(self): time.sleep(0.01) break - Settings.get_instance().set_value(SettingsConstants.SETTING__QR_BRIGHTNESS, self.qr_brightness.cur_count) + Settings.get_instance().set_value(SettingsConstants.SETTING__QR_BRIGHTNESS, self.qr_brightness.cur_value) @@ -894,7 +891,7 @@ def render_border(color, width): ) try: - while self.keep_running: + while not self.event.wait(timeout=0.05): # Target ~10fps with screen.renderer.lock: # Ramp the edges from a darker version out to full color inhale_scalar = inhale_factor * int(255/inhale_max) @@ -924,9 +921,6 @@ def render_border(color, width): inhale_factor = 1 inhale_factor += inhale_step - # Target ~10fps - time.sleep(0.05) - except KeyboardInterrupt as e: self.stop() raise e diff --git a/src/seedsigner/gui/screens/seed_screens.py b/src/seedsigner/gui/screens/seed_screens.py index 9fe35c1c..557a6e46 100644 --- a/src/seedsigner/gui/screens/seed_screens.py +++ b/src/seedsigner/gui/screens/seed_screens.py @@ -7,7 +7,7 @@ from PIL import Image, ImageDraw, ImageFilter from seedsigner.gui.renderer import Renderer from seedsigner.helpers.qr import QR -from seedsigner.models.threads import BaseThread, ThreadsafeCounter +from seedsigner.models.threads import BaseThread, ThreadsafeCounter, ThreadsafeVar from .screen import RET_CODE__BACK_BUTTON, BaseScreen, BaseTopNavScreen, ButtonListScreen, KeyboardScreen, WarningEdgesMixin from ..components import (Button, FontAwesomeIconConstants, Fonts, FormattedAddress, IconButton, @@ -1370,8 +1370,8 @@ class SeedAddressVerificationScreen(ButtonListScreen): sig_type: str = None network: str = None is_mainnet: bool = None - threadsafe_counter: ThreadsafeCounter = None - verified_index: ThreadsafeCounter = None + cur_addr_index: ThreadsafeCounter = None + verified_index: ThreadsafeVar[int] = None def __post_init__(self): @@ -1404,34 +1404,33 @@ def __post_init__(self): self.threads.append(SeedAddressVerificationScreen.ProgressThread( renderer=self.renderer, screen_y=self.components[-1].screen_y + self.components[-1].height + GUIConstants.COMPONENT_PADDING, - threadsafe_counter=self.threadsafe_counter, + cur_addr_index=self.cur_addr_index, verified_index=self.verified_index, )) def _run_callback(self): # Exit the screen on success via a non-None value - print(f"verified_index: {self.verified_index.cur_count}") - if self.verified_index.cur_count is not None: - print("Screen callback returning success!") + if self.verified_index.cur_value is not None: self.threads[-1].stop() - while self.threads[-1].is_alive(): - time.sleep(0.01) + + # Wait for the thread to exit + self.threads[-1].join() return 1 class ProgressThread(BaseThread): - def __init__(self, renderer: Renderer, screen_y: int, threadsafe_counter: ThreadsafeCounter, verified_index: ThreadsafeCounter): + def __init__(self, renderer: Renderer, screen_y: int, cur_addr_index: ThreadsafeCounter, verified_index: ThreadsafeVar[int]): self.renderer = renderer self.screen_y = screen_y - self.threadsafe_counter = threadsafe_counter + self.cur_addr_index = cur_addr_index self.verified_index = verified_index super().__init__() def run(self): - while self.keep_running: - if self.verified_index.cur_count is not None: + while not self.event.wait(timeout=0.1): + if self.verified_index.cur_value is not None: # This thread will detect the success state while its parent Screen # holds in its `wait_for`. Have to trigger a hw_input event to break # the Screen._run out of the `wait_for` state. The Screen will then @@ -1440,7 +1439,7 @@ def run(self): return textarea = TextArea( - text=f"Checking address {self.threadsafe_counter.cur_count}", + text=f"Checking address {self.cur_addr_index.cur_value}", font_name=GUIConstants.BODY_FONT_NAME, font_size=GUIConstants.BODY_FONT_SIZE, screen_y=self.screen_y @@ -1450,8 +1449,6 @@ def run(self): textarea.render() self.renderer.show_image() - time.sleep(0.1) - @dataclass diff --git a/src/seedsigner/gui/toast.py b/src/seedsigner/gui/toast.py index 7a07f30b..eef6abce 100644 --- a/src/seedsigner/gui/toast.py +++ b/src/seedsigner/gui/toast.py @@ -1,6 +1,7 @@ import time from dataclasses import dataclass from seedsigner.gui.components import BaseComponent, GUIConstants, Icon, SeedSignerIconConstants, TextArea +from seedsigner.hardware.microsd import MicroSD from seedsigner.models.threads import BaseThread @@ -70,7 +71,7 @@ class BaseToastOverlayManagerThread(BaseThread): manager thread that the Controller will use to coordinate handing off resources between competing toasts, the screensaver, and the current underlying Screen. - Controller should set BaseThread.keep_running = False to terminate the toast when it + Controller should call the thread's stop() to terminate the toast when it needs to be removed or replaced. Controller should set toggle_renderer_lock = True to make the toast temporarily @@ -109,11 +110,6 @@ def instantiate_toast(self) -> ToastOverlay: raise Exception("Must be implemented by subclass") - def should_keep_running(self) -> bool: - """ Placeholder for custom exit conditions """ - return True - - def toggle_renderer_lock(self): self._toggle_renderer_lock = True @@ -137,8 +133,11 @@ def run(self): has_rendered = False previous_screen_state = None - while self.keep_running and self.should_keep_running(): - if self.hw_inputs.has_any_input(): + while not self.event.wait(timeout=0.1): + if not MicroSD.get_instance().is_inserted: + break + + elif self.hw_inputs.has_any_input(): # User has pressed a button, hide the toast print(f"{self.__class__.__name__}: Exiting due to user input") break @@ -169,9 +168,6 @@ def run(self): print(f"{self.__class__.__name__}: Hiding toast") break - # Free up cpu resources for main thread - time.sleep(0.1) - finally: print(f"{self.__class__.__name__}: exiting") if has_rendered and self.renderer.lock.locked(): @@ -203,12 +199,6 @@ def instantiate_toast(self) -> ToastOverlay: ) - def should_keep_running(self) -> bool: - """ Custom exit condition: keep running until the SD card is removed """ - from seedsigner.hardware.microsd import MicroSD - return MicroSD.get_instance().is_inserted - - class SDCardStateChangeToastManagerThread(BaseToastOverlayManagerThread): def __init__(self, action: str, *args, **kwargs): diff --git a/src/seedsigner/hardware/microsd.py b/src/seedsigner/hardware/microsd.py index 26065c03..9aae4c09 100644 --- a/src/seedsigner/hardware/microsd.py +++ b/src/seedsigner/hardware/microsd.py @@ -60,12 +60,10 @@ def run(self): os.mkfifo(self.FIFO_PATH, self.FIFO_MODE) - while self.keep_running: + while not self.event.wait(timeout=0.1): with open(self.FIFO_PATH) as fifo: action = fifo.read() print(f"fifo message: {action}") Settings.handle_microsd_state_change(action=action) Controller.get_instance().activate_toast(SDCardStateChangeToastManagerThread(action=action)) - - time.sleep(0.1) diff --git a/src/seedsigner/models/threads.py b/src/seedsigner/models/threads.py index 0fe5651c..3091da72 100644 --- a/src/seedsigner/models/threads.py +++ b/src/seedsigner/models/threads.py @@ -1,46 +1,66 @@ +import copy import logging -from threading import Thread, Lock +from threading import Event, Thread, Lock +from typing import Generic, TypeVar logger = logging.getLogger(__name__) +T = TypeVar("T") + + + class BaseThread(Thread): - def __init__(self): + def __init__(self, event: Event = None): super().__init__(daemon=True) - - def start(self): - logger.debug(f"{self.__class__.__name__} STARTING") - self.keep_running = True - super().start() + if event is None: + # Have to instantiate a default Event here; if we do it as a default value + # for the kwarg, we inadvertently end up re-using the same Event instance. + event = Event() + self.event = event + def stop(self): - logger.debug(f"{self.__class__.__name__} EXITING") - self.keep_running = False - + # Set the thread's Event; the thread's execution loop will exit on its next + # iteration. + self.event.set() + + def run(self): - while self.keep_running: - # Do something - raise Exception(f"Must implement run() in {self.__class__.__name__}") + """ + while not self.event.is_set(): + # Do something + """ + raise Exception(f"Must implement run() in {self.__class__.__name__}") -class ThreadsafeCounter: - def __init__(self, initial_value: int = 0): - self.count = initial_value +class ThreadsafeVar(Generic[T]): + def __init__(self, initial_value: T = None): + self.data: T = initial_value self._lock = Lock() - + + @property - def cur_count(self): + def cur_value(self) -> T: # Reads don't require the lock - return self.count + return copy.copy(self.data) - def increment(self, step: int = 1): + + def set_value(self, value: T): # Updates must be locked with self._lock: - self.count += step - - def set_value(self, value: int): - with self._lock: - self.count = value + self.data = value + +class ThreadsafeCounter(ThreadsafeVar[int]): + def __init__(self, initial_value: int = 0): + # Must initialize to some starting int value + super().__init__(initial_value) + + + def increment(self, step: int = 1): + # Updates must be locked + with self._lock: + self.data += step diff --git a/src/seedsigner/views/seed_views.py b/src/seedsigner/views/seed_views.py index afd70e10..cd6a8179 100644 --- a/src/seedsigner/views/seed_views.py +++ b/src/seedsigner/views/seed_views.py @@ -22,7 +22,7 @@ from seedsigner.models.seed import InvalidSeedException, Seed from seedsigner.models.settings import Settings, SettingsConstants from seedsigner.models.settings_definition import SettingsDefinition -from seedsigner.models.threads import BaseThread, ThreadsafeCounter +from seedsigner.models.threads import BaseThread, ThreadsafeCounter, ThreadsafeVar from seedsigner.views.view import NotYetImplementedView, OptionDisabledView, View, Destination, BackStackView, MainMenuView @@ -1665,11 +1665,11 @@ def __init__(self, seed_num: int = None): # The ThreadsafeCounter will be shared by the brute-force thread to keep track of # its current addr index number and the Screen to display its progress and # respond to UI requests to jump the index ahead. - self.threadsafe_counter = ThreadsafeCounter() + self.cur_addr_index = ThreadsafeCounter() # Shared coordination var so the display thread can detect success - self.verified_index = ThreadsafeCounter(initial_value=None) - self.verified_index_is_change = ThreadsafeCounter(initial_value=None) + self.verified_index = ThreadsafeVar[int]() + self.verified_index_is_change = ThreadsafeVar[bool]() # Create the brute-force calculation thread that will run in the background self.addr_verification_thread = self.BruteForceAddressVerificationThread( @@ -1679,7 +1679,7 @@ def __init__(self, seed_num: int = None): script_type=self.script_type, embit_network=embit_network, derivation_path=self.derivation_path, - threadsafe_counter=self.threadsafe_counter, + cur_addr_index=self.cur_addr_index, verified_index=self.verified_index, verified_index_is_change=self.verified_index_is_change, ) @@ -1708,49 +1708,56 @@ def run(self): # and resume displaying the screen. User won't even notice that the Screen is # being re-constructed. while True: - selected_menu_num = seed_screens.SeedAddressVerificationScreen( + selected_menu_num = self.run_screen( + seed_screens.SeedAddressVerificationScreen, address=self.address, derivation_path=self.derivation_path, script_type=script_type_display, sig_type=sig_type_display, network=network_display, is_mainnet=network_display == mainnet, - threadsafe_counter=self.threadsafe_counter, + cur_addr_index=self.cur_addr_index, verified_index=self.verified_index, button_data=button_data, - ).display() + ) - if self.verified_index.cur_count is not None: + if selected_menu_num is None: + # Only occurs during the test suite's flow tests since it doesn't + # actually run the Screen. The test must wait for the brute force thread + # to complete its work. + # TODO: Add an IS_TESTING env var to ensure that we're in the test suite? + self.addr_verification_thread.join() + break + + if self.verified_index.cur_value is not None: break if selected_menu_num == RET_CODE__BACK_BUTTON: break if button_data[selected_menu_num] == SKIP_10: - self.threadsafe_counter.increment(10) + self.cur_addr_index.increment(10) elif button_data[selected_menu_num] == CANCEL: break - if self.verified_index.cur_count is not None: + if self.verified_index.cur_value is not None: # Successfully verified the addr; update the data - self.controller.unverified_address["verified_index"] = self.verified_index.cur_count - self.controller.unverified_address["verified_index_is_change"] = self.verified_index_is_change.cur_count == 1 + self.controller.unverified_address["verified_index"] = self.verified_index.cur_value + self.controller.unverified_address["verified_index_is_change"] = self.verified_index_is_change.cur_value == 1 return Destination(AddressVerificationSuccessView, view_args=dict(seed_num=self.seed_num)) else: # Halt the thread if the user gave up (will already be stopped if it verified the # target addr). self.addr_verification_thread.stop() - while self.addr_verification_thread.is_alive(): - time.sleep(0.01) return Destination(MainMenuView) class BruteForceAddressVerificationThread(BaseThread): - def __init__(self, address: str, seed: Seed, descriptor: Descriptor, script_type: str, embit_network: str, derivation_path: str, threadsafe_counter: ThreadsafeCounter, verified_index: ThreadsafeCounter, verified_index_is_change: ThreadsafeCounter): + def __init__(self, address: str, seed: Seed, descriptor: Descriptor, script_type: str, embit_network: str, derivation_path: str, cur_addr_index: ThreadsafeCounter, verified_index: ThreadsafeVar[int], verified_index_is_change: ThreadsafeVar[bool]): """ Either seed or descriptor will be None """ @@ -1761,7 +1768,7 @@ def __init__(self, address: str, seed: Seed, descriptor: Descriptor, script_type self.script_type = script_type self.embit_network = embit_network self.derivation_path = derivation_path - self.threadsafe_counter = threadsafe_counter + self.cur_addr_index = cur_addr_index self.verified_index = verified_index self.verified_index_is_change = verified_index_is_change @@ -1770,11 +1777,11 @@ def __init__(self, address: str, seed: Seed, descriptor: Descriptor, script_type def run(self): - while self.keep_running: - if self.threadsafe_counter.cur_count % 10 == 0: - print(f"Incremented to {self.threadsafe_counter.cur_count}") + while not self.event.is_set(): + if self.cur_addr_index.cur_value % 10 == 0: + print(f"Incremented to {self.cur_addr_index.cur_value}") - i = self.threadsafe_counter.cur_count + i = self.cur_addr_index.cur_value if self.descriptor: receive_address = embit_utils.get_multisig_address(descriptor=self.descriptor, index=i, is_change=False, embit_network=self.embit_network) @@ -1786,19 +1793,17 @@ def run(self): if self.address == receive_address: self.verified_index.set_value(i) - self.verified_index_is_change.set_value(0) - self.keep_running = False + self.verified_index_is_change.set_value(False) break elif self.address == change_address: self.verified_index.set_value(i) - self.verified_index_is_change.set_value(1) - self.keep_running = False + self.verified_index_is_change.set_value(True) break # Increment our index counter - self.threadsafe_counter.increment() - + self.cur_addr_index.increment() + class AddressVerificationSuccessView(View): @@ -1820,11 +1825,12 @@ def run(self): else: source = f"seed {self.seed.get_fingerprint()}" - LargeIconStatusScreen( + self.run_screen( + LargeIconStatusScreen, status_headline="Address Verified", text=f"""{address[:7]} = {source}'s {"change" if verified_index_is_change else "receive"} address #{verified_index}.""", show_back_button=False, - ).display() + ) return Destination(MainMenuView) diff --git a/src/seedsigner/views/view.py b/src/seedsigner/views/view.py index a5c4800c..e6aa0b04 100644 --- a/src/seedsigner/views/view.py +++ b/src/seedsigner/views/view.py @@ -281,9 +281,8 @@ class PowerOffThread(BaseThread): def run(self): import time from subprocess import call - while self.keep_running: - time.sleep(5) - call("sudo shutdown --poweroff now", shell=True) + time.sleep(5) + call("sudo shutdown --poweroff now", shell=True) diff --git a/tests/test_flows_seed.py b/tests/test_flows_seed.py index e71ad0b5..ad33a241 100644 --- a/tests/test_flows_seed.py +++ b/tests/test_flows_seed.py @@ -312,6 +312,38 @@ def test_discard_seed_flow(self): ) + def test_addressverification_flow(self): + """ + Scanning an address QR from Home should enter the Address Verification flow, + including prompting to select or load a seed. Brute-force address verification + should proceed in a child thread until found. + """ + # Test data generated by bitcoiner.guide/seed + seed = Seed(mnemonic="able ignore obey define rely seminar icon employ polar alert scatter celery".split()) + target_address = "bc1q0w20252pfn6dch6ag0jceseymd8hnqlhv9y89d" + expected_index = 14 + + self.controller.storage.set_pending_seed(seed) + self.controller.storage.finalize_pending_seed() + + def load_address_into_decoder(view: scan_views.ScanView): + view.decoder.add_data(target_address) + + def verify_results(view: View): + assert self.controller.unverified_address["address"] == target_address + assert self.controller.unverified_address["verified_index"] == expected_index + + self.run_sequence([ + FlowStep(MainMenuView, button_data_selection=MainMenuView.SCAN), + FlowStep(scan_views.ScanView, before_run=load_address_into_decoder), # simulate read address QR + FlowStep(seed_views.AddressVerificationStartView, is_redirect=True), + FlowStep(seed_views.SeedSelectSeedView, screen_return_value=0), # ret 1st onboard seed + FlowStep(seed_views.SeedAddressVerificationView), + FlowStep(seed_views.AddressVerificationSuccessView, before_run=verify_results), + FlowStep(MainMenuView) + ]) + + class TestMessageSigningFlows(FlowTest): MAINNET_DERIVATION_PATH = "m/84h/0h/0h/0/0" @@ -465,8 +497,6 @@ def expect_network_mismatch_error(load_message: Callable): expect_network_mismatch_error(self.load_short_message_into_decoder) - - def test_sign_message_option_disabled(self): """ Should redirect to OptionDisabledView if a `signmessage` QR is scanned with diff --git a/tests/test_seed.py b/tests/test_seed.py index a7e082ee..c1d4e86d 100644 --- a/tests/test_seed.py +++ b/tests/test_seed.py @@ -1,41 +1,79 @@ +# Must import test base before the Controller +import time +from base import BaseTest + from seedsigner.models.seed import Seed from seedsigner.models.settings import SettingsConstants +from seedsigner.models.threads import ThreadsafeCounter, ThreadsafeVar +from seedsigner.views.seed_views import SeedAddressVerificationView def test_seed(): - seed = Seed(mnemonic="obscure bone gas open exotic abuse virus bunker shuffle nasty ship dash".split()) - - assert seed.seed_bytes == b'q\xb3\xd1i\x0c\x9b\x9b\xdf\xa7\xd9\xd97H\xa8,\xa7\xd9>\xeck\xc2\xf5ND?, \x88-\x07\x9aa\xc5\xee\xb7\xbf\xc4x\xd6\x07 X\xb6}?M\xaa\x05\xa6\xa7(>\xbf\x03\xb0\x9d\xef\xed":\xdf\x88w7' - - assert seed.mnemonic_str == "obscure bone gas open exotic abuse virus bunker shuffle nasty ship dash" - - assert seed.passphrase == "" - - # TODO: Not yet supported in new implementation - # seed.set_wordlist_language_code("es") - - # assert seed.mnemonic_str == "natural ayuda futuro nivel espejo abuelo vago bien repetir moreno relevo conga" - - # seed.set_wordlist_language_code(SettingsConstants.WORDLIST_LANGUAGE__ENGLISH) - - # seed.mnemonic_str = "height demise useless trap grow lion found off key clown transfer enroll" - - # assert seed.mnemonic_str == "height demise useless trap grow lion found off key clown transfer enroll" - - # # TODO: Not yet supported in new implementation - # seed.set_wordlist_language_code("es") - - # assert seed.mnemonic_str == "hebilla cría truco tigre gris llenar folio negocio laico casa tieso eludir" - - # seed.set_passphrase("test") - - # assert seed.seed_bytes == b'\xdd\r\xcb\x0b V\xb4@\xee+\x01`\xabem\xc1B\xfd\x8fba0\xab;[\xab\xc9\xf9\xba[F\x0c5,\x7fd8\xebI\x90"\xb8\x86C\x821\x01\xdb\xbe\xf3\xbc\x1cBH"%\x18\xc2{\x04\x08a]\xa5' - - # assert seed.passphrase == "test" - - - - - \ No newline at end of file + seed = Seed(mnemonic="obscure bone gas open exotic abuse virus bunker shuffle nasty ship dash".split()) + + assert seed.seed_bytes == b'q\xb3\xd1i\x0c\x9b\x9b\xdf\xa7\xd9\xd97H\xa8,\xa7\xd9>\xeck\xc2\xf5ND?, \x88-\x07\x9aa\xc5\xee\xb7\xbf\xc4x\xd6\x07 X\xb6}?M\xaa\x05\xa6\xa7(>\xbf\x03\xb0\x9d\xef\xed":\xdf\x88w7' + + assert seed.mnemonic_str == "obscure bone gas open exotic abuse virus bunker shuffle nasty ship dash" + + assert seed.passphrase == "" + + # TODO: Not yet supported in new implementation + # seed.set_wordlist_language_code("es") + + # assert seed.mnemonic_str == "natural ayuda futuro nivel espejo abuelo vago bien repetir moreno relevo conga" + + # seed.set_wordlist_language_code(SettingsConstants.WORDLIST_LANGUAGE__ENGLISH) + + # seed.mnemonic_str = "height demise useless trap grow lion found off key clown transfer enroll" + + # assert seed.mnemonic_str == "height demise useless trap grow lion found off key clown transfer enroll" + + # # TODO: Not yet supported in new implementation + # seed.set_wordlist_language_code("es") + + # assert seed.mnemonic_str == "hebilla cría truco tigre gris llenar folio negocio laico casa tieso eludir" + + # seed.set_passphrase("test") + + # assert seed.seed_bytes == b'\xdd\r\xcb\x0b V\xb4@\xee+\x01`\xabem\xc1B\xfd\x8fba0\xab;[\xab\xc9\xf9\xba[F\x0c5,\x7fd8\xebI\x90"\xb8\x86C\x821\x01\xdb\xbe\xf3\xbc\x1cBH"%\x18\xc2{\x04\x08a]\xa5' + + # assert seed.passphrase == "test" + + + +class TestBruteForceAddressVerificationThread(BaseTest): + def test_brute_force(self): + """ + Ensure that the child thread that does the brute force address verification finds + the expected result. + """ + # TODO: test all supported address types (e.g. nested segwit) + # Test data generated by bitcoiner.guide/seed + seed = Seed(mnemonic="able ignore obey define rely seminar icon employ polar alert scatter celery".split()) + target_address = "bc1qy3y3zq7rclp9fwds2z95w9ru3r5pkyfhchcqgq" + expected_index = 21 + + cur_addr_index = ThreadsafeCounter() + verified_index = ThreadsafeVar[int]() + verified_index_is_change = ThreadsafeVar[bool]() + + brute_force_thread = SeedAddressVerificationView.BruteForceAddressVerificationThread( + address=target_address, + seed=seed, + descriptor=None, + script_type=SettingsConstants.NATIVE_SEGWIT, + embit_network=SettingsConstants.map_network_to_embit(SettingsConstants.MAINNET), + derivation_path="m/84'/0'/0'", + cur_addr_index=cur_addr_index, + verified_index=verified_index, + verified_index_is_change=verified_index_is_change + ) + brute_force_thread.start() + + # Block current test until child thread completes + brute_force_thread.join() + + assert verified_index.cur_value == expected_index + assert verified_index_is_change.cur_value is False diff --git a/tests/test_threads.py b/tests/test_threads.py new file mode 100644 index 00000000..d9db8bfc --- /dev/null +++ b/tests/test_threads.py @@ -0,0 +1,95 @@ +from threading import Event +import time +from seedsigner.models.threads import T, BaseThread, ThreadsafeVar + + + +class SimpleThreadTester(BaseThread): + def __init__(self, threadsafe_var: ThreadsafeVar[T] = None, exit_when_equal: T = None): + super().__init__() + self.threadsafe_var = threadsafe_var + self.exit_when_equal = exit_when_equal + + + def run(self): + while not self.event.wait(timeout=0.1): + if self.threadsafe_var.cur_value == self.exit_when_equal: + break + + print("Thread exiting itself") + + + +class TestThreadClass: + def test_thread_event_exit(self): + threadsafe_var = ThreadsafeVar[int]() + exit_when_equal = 100 + test_thread = SimpleThreadTester(threadsafe_var=threadsafe_var, exit_when_equal=exit_when_equal) + test_thread.start() + + assert test_thread.is_alive() + + # Now signal the thread to cancel itself + test_thread.stop() + test_thread.join() + assert not test_thread.is_alive() + + + +class TestThreadsafeVar: + def test_threadsafevar_generic_type_get_set(self): + threadsafe_var = ThreadsafeVar[int]() + assert threadsafe_var.cur_value is None + threadsafe_var.set_value(21) + assert threadsafe_var.cur_value == 21 + + threadsafe_var = ThreadsafeVar[bool]() + assert threadsafe_var.cur_value is None + threadsafe_var.set_value(True) + assert threadsafe_var.cur_value is True + + threadsafe_var = ThreadsafeVar[str]() + assert threadsafe_var.cur_value is None + threadsafe_var.set_value("satoshi") + assert threadsafe_var.cur_value is "satoshi" + + + def test_threadsafevar_int(self): + threadsafe_var = ThreadsafeVar[int]() + exit_when_equal = 10 + test_thread = SimpleThreadTester(threadsafe_var=threadsafe_var, exit_when_equal=exit_when_equal) + test_thread.start() + + while not test_thread.is_alive(): + time.sleep(0.1) + + for i in range(1, exit_when_equal): + threadsafe_var.set_value(i) + assert threadsafe_var.cur_value != exit_when_equal + assert test_thread.is_alive() + + threadsafe_var.set_value(exit_when_equal) + assert threadsafe_var.cur_value == exit_when_equal + + # Give the thread a moment to end itself + test_thread.join() + + assert not test_thread.is_alive() + + + def test_threadsafevar_bool(self): + threadsafe_var = ThreadsafeVar[bool]() + exit_when_equal = True + test_thread = SimpleThreadTester(threadsafe_var=threadsafe_var, exit_when_equal=exit_when_equal) + test_thread.start() + + threadsafe_var.set_value(False) + assert test_thread.is_alive() + + threadsafe_var.set_value(exit_when_equal) + + # Give the thread a moment to end itself + test_thread.join() + + assert not test_thread.is_alive() + \ No newline at end of file