From 4c5726571e21902813d0117d2c94e5a853308b19 Mon Sep 17 00:00:00 2001 From: GenSpark AI Developer Date: Mon, 15 Jun 2026 14:29:04 +0000 Subject: [PATCH] feat(net): invio/ping/download verso Will-Executor in parallelo (anti-freeze) Problema: i server Will-Executor venivano contattati in sequenza e, su timeout, send_request riprovava 10x con sleep 3s (~130s per server morto). Un solo server irraggiungibile bloccava l'intera operazione (impallamento UI). Soluzione: - ThreadPoolExecutor: ping e push ora in parallelo (tempo ~= server piu' lento, non la somma). Un server morto non blocca piu' gli altri. - Fast-fail per operazioni interattive (ping/info/download): max_retries=0, niente retry-storm. - Feedback live: callback on_each() aggiorna il dialog server-per-server (thread-safe via pyqtSignal di BalWaitingDialog.update). - Push transazioni: parallelo ma con retry per-server mantenuti (no perdita tx). File: - bal/core/willexecutors.py: send_request(+max_retries,+retry_sleep), get_info_task fast-fail, NEW ping_servers_parallel(), push_transactions_parallel(), DEFAULT_TIMEOUT=5. - bal/gui/qt/window.py: ping_willexecutors_task + push_transactions_to_willexecutors riscritti su helper paralleli con feedback live; fetch_will_executors_list fast-fail. - bal/core/util.py: BUGFIX get_value_amount usava in_output (bool) invece di din_output (tupla) -> TypeError. Scoperto dai test ufficiali Gitea. Test (contro il codice refactor): - pytest tests/ ufficiali: 117 core + 65 gui = 182 passed. - smoke/external_zip/windows_overflow/gui_fixes: OK. - parallel_ping_test (nuovo): 0.50s per 8 server vs ~4.00s sequenziale. - ruff: nessun nuovo problema introdotto (codice nuovo PEP8-compliant). Aggiunti i test ufficiali del repo Gitea + REPORT_NETWORKING_PARALLELO.md. --- REPORT_NETWORKING_PARALLELO.md | 164 ++++++++++++ bal/core/util.py | 2 +- bal/core/willexecutors.py | 143 +++++++++- bal/gui/qt/window.py | 145 ++++++---- tests/parallel_ping_test.py | 130 +++++++++ tests/test_core_heirs.py | 266 +++++++++++++++++++ tests/test_core_heirs_extra.py | 182 +++++++++++++ tests/test_core_plugin_base.py | 259 ++++++++++++++++++ tests/test_core_util.py | 470 +++++++++++++++++++++++++++++++++ tests/test_core_will.py | 272 +++++++++++++++++++ tests/test_core_will_extra.py | 167 ++++++++++++ tests/test_gui_calendar.py | 142 ++++++++++ tests/test_gui_common.py | 102 +++++++ tests/test_gui_theme.py | 100 +++++++ tests/test_gui_widgets.py | 255 ++++++++++++++++++ tests/test_gui_window_utils.py | 103 ++++++++ 16 files changed, 2843 insertions(+), 59 deletions(-) create mode 100644 REPORT_NETWORKING_PARALLELO.md create mode 100644 tests/parallel_ping_test.py create mode 100644 tests/test_core_heirs.py create mode 100644 tests/test_core_heirs_extra.py create mode 100644 tests/test_core_plugin_base.py create mode 100644 tests/test_core_util.py create mode 100644 tests/test_core_will.py create mode 100644 tests/test_core_will_extra.py create mode 100644 tests/test_gui_calendar.py create mode 100644 tests/test_gui_common.py create mode 100644 tests/test_gui_theme.py create mode 100644 tests/test_gui_widgets.py create mode 100644 tests/test_gui_window_utils.py diff --git a/REPORT_NETWORKING_PARALLELO.md b/REPORT_NETWORKING_PARALLELO.md new file mode 100644 index 0000000..6cbfbc0 --- /dev/null +++ b/REPORT_NETWORKING_PARALLELO.md @@ -0,0 +1,164 @@ +# Report tecnico — Networking parallelo (anti-freeze Will-Executor) + +**Destinatario:** programmatore esterno / manutentore del plugin +**Autore:** refactoring AI (lavoro su GitHub `Bitcoin-after-life/test`) +**Data:** 2026-06-15 +**Branch:** `feature/networking-parallelo` +**Repository Gitea privato `kaibot/bal-plugin-ai`: NON modificato** (per richiesta esplicita). + +--- + +## 1. Problema + +Quando il plugin contatta i server Will-Executor (invio transazioni, ping/aggiornamento +eredità, download lista), lo fa **in sequenza**. Se un server non risponde, il thread +resta bloccato sui timeout di connessione e, peggio, sui **retry**: + +- `send_request` riprovava fino a **10 volte** con `time.sleep(3)` ad ogni timeout + → circa **130 secondi per ogni server irraggiungibile**, sommati uno dopo l'altro. + +Conseguenze: +- Con pochi server già si avverte; con **20 server** diventa ingestibile. +- L'utente vede "Rimani in attesa — Non risponde" senza capire cosa succede. +- Un singolo server morto blocca l'intera operazione. + +--- + +## 2. Soluzione (in sintesi) + +1. **Parallelismo** con `ThreadPoolExecutor`: i server vengono contattati + contemporaneamente. Il tempo totale ≈ server **più lento**, non la **somma**. +2. **Fast-fail** per le operazioni interattive (ping/info/download): niente retry-storm, + un solo timeout breve e il server viene marcato "KO". +3. **Feedback live**: callback `on_each(...)` che aggiorna la finestra di attesa + server-per-server, in modo thread-safe. +4. **Push transazioni**: parallelo, ma mantiene i retry per ogni singolo server + (una transazione reale non deve andare persa per un hiccup transitorio). + +### Perché è thread-safe +`Network.send_http_on_proxy()` usa `asyncio.run_coroutine_threadsafe(coro, loop)` e poi +`coro.result()`: ogni chiamata schedula la propria coroutine sullo stesso loop asyncio +condiviso di Electrum e blocca **solo il proprio worker thread**. Più chiamate +concorrenti sono quindi sicure → `ThreadPoolExecutor` dà vero parallelismo. + +Gli aggiornamenti UI passano per `BalWaitingDialog.update()` che emette un +`pyqtSignal` → marshalling automatico sul thread GUI. I callback dai worker thread +possono quindi aggiornare il dialog in sicurezza. + +--- + +## 3. File modificati (tutto su GitHub `Bitcoin-after-life/test`, branch `feature/networking-parallelo`) + +### 3.1 `bal/core/willexecutors.py` + +**`send_request(...)`** — aggiunti due parametri keyword-only: +```python +def send_request(method, url, data=None, *, timeout=10, handle_response=None, + count_reply=0, max_retries=10, retry_sleep=3): +``` +- `max_retries` / `retry_sleep` controllano i retry sui timeout. +- **Default invariato** (`10` / `3s`) → il push critico mantiene il comportamento storico. +- Le operazioni interattive passano `max_retries=0` → fast-fail. + +**`get_info_task(...)`** — fast-fail di default: +```python +def get_info_task(url, willexecutor, *, timeout=DEFAULT_TIMEOUT, + max_retries=0, retry_sleep=0): +``` +- Se la risposta non è un `dict` (timeout/vuoto) → `status="KO"`. + +**NUOVO `ping_servers_parallel(willexecutors, *, on_each=None, max_workers=8, timeout=DEFAULT_TIMEOUT)`** +- `ThreadPoolExecutor` + `as_completed`; un worker per server (`_ping_one`). +- Muta `willexecutors` in place (come il vecchio `ping_servers`). +- Invoca `on_each(url, we, ok)` man mano che arrivano i risultati. +- Un worker che esplode non blocca gli altri (try/except difensivo). + +**NUOVO `push_transactions_parallel(willexecutors, *, on_each=None, max_workers=8)`** +- Push in parallelo solo verso le voci con chiave `"txs"`. +- Ogni server mantiene i propri retry (`push_transactions_to_willexecutor`). +- `on_each(url, we, ok, exc)`; raccoglie `AlreadyPresentException` separatamente. +- Ritorna `{url: (ok, exc)}`. + +**`DEFAULT_TIMEOUT = 5`** (costante a livello di modulo). + +### 3.2 `bal/gui/qt/window.py` + +- **`ping_willexecutors_task(self, wes)`** riscritto su `ping_servers_parallel(...)` + con feedback live (set `pinged`/`failed`, `get_title()` mostra "Ok"/"Ko"/"waiting..."). +- **`push_transactions_to_willexecutors(self, force=False)`** riscritto su + `push_transactions_parallel(...)`. `on_each` fa book-keeping + update UI thread-safe; + i server "already present" sono raccolti in `already_present[]` e il loro + `check_transaction` viene eseguito dopo, nel task thread (logica di check originale intatta). +- **`fetch_will_executors_list(...)`** download fast-fail: + `send_request("get", url, timeout=10, max_retries=1, retry_sleep=1)`. + +### 3.3 `bal/core/util.py` — BUGFIX (regressione pre-esistente) + +In `get_value_amount` (riga 324) era stato erroneamente usato `Util.in_output(...)` +(ritorna `bool`) al posto di `Util.din_output(...)` (ritorna la tupla +`(same_amount, same_address)`), causando: +``` +TypeError: cannot unpack non-iterable bool object +``` +**Corretto** ripristinando `din_output`. Bug scoperto eseguendo i test ufficiali del +repo Gitea (`tests/test_core_util.py::test_get_value_amount`). + +--- + +## 4. Verifica (ruff + test ufficiali) + +### 4.1 ruff (lint / PEP8) +- `ruff check` sul codice nuovo: **nessun nuovo problema** introdotto + (i `F403/F405/F401` presenti derivano dal pattern `from .common import *` + dell'originale; conteggio identico HEAD vs working tree: 121 = 121). +- Le funzioni parallele nuove rispettano il limite di 88 caratteri (0 `E501`). +- `ruff check tests/parallel_ping_test.py` → **All checks passed**. + +### 4.2 Test ufficiali del repo Gitea `kaibot/bal-plugin-ai/tests` +Eseguiti contro il codice refactorizzato (con le modifiche networking): + +| Suite | Esito | +|-------|-------| +| `test_core_*` (pytest) | **117 passed** | +| `test_gui_*` (pytest) | **65 passed** | +| `smoke_test.py` | OK | +| `external_zip_test.py` | OK | +| `windows_overflow_test.py` | OK | +| `gui_fixes_test.py` | OK | +| `parallel_ping_test.py` (nuovo) | OK — `0.50s` per 8 server (sequenziale ~`4.00s`) | + +Comandi (come da README): +```bash +QT_QPA_PLATFORM=offscreen PYTHONPATH= \ + python3 -m pytest tests/ -q +QT_QPA_PLATFORM=offscreen PYTHONPATH= \ + python3 tests/smoke_test.py electrum.plugins.bal +QT_QPA_PLATFORM=offscreen PYTHONPATH= \ + python3 tests/external_zip_test.py bal-electrum-plugin.zip +``` + +--- + +## 5. Note di integrazione / rischi + +- **Nessuna modifica al protocollo server**: solo il *come* (parallelo) e il *quando* + (retry) delle chiamate cambia, non i payload. +- **Push transazioni**: i retry per-server sono mantenuti apposta, per non perdere + una transazione reale per un hiccup. Solo il ping/info/download usa fast-fail. +- **`max_workers=8`** è prudente; con molti server (es. 20) si può alzare, ma 8 + worker già abbattono il tempo totale al server più lento. +- **Thread/UI**: tutti gli aggiornamenti UI dai worker passano per + `BalWaitingDialog.update()` (pyqtSignal) → safe. Non toccare quel canale. +- **Compatibilità**: firme retro-compatibili (i nuovi parametri sono keyword-only + con default che preservano il vecchio comportamento). + +--- + +## 6. Come provare + +1. Installare lo zip `bal-electrum-plugin.zip` (Tools → Plugins → install from file). +2. Configurare più Will-Executor, includendone **almeno uno irraggiungibile**. +3. Lanciare invio transazioni / ping: il dialog mostra lo stato server-per-server + e **non resta più bloccato** sul server morto. + +SHA-256 dello zip stampato da `build_zip.py` a fine build (verificare l'integrità). diff --git a/bal/core/util.py b/bal/core/util.py index 2ff6749..32d7ec1 100644 --- a/bal/core/util.py +++ b/bal/core/util.py @@ -321,7 +321,7 @@ class Util: value_amount = 0 for outa in outputsa: - same_amount, same_address = Util.in_output(outa, txb.outputs()) + same_amount, same_address = Util.din_output(outa, txb.outputs()) if not (same_amount or same_address): return False if same_amount and same_address: diff --git a/bal/core/willexecutors.py b/bal/core/willexecutors.py index e025ac8..4231766 100644 --- a/bal/core/willexecutors.py +++ b/bal/core/willexecutors.py @@ -145,8 +145,21 @@ class Willexecutors: @staticmethod def send_request( - method, url, data=None, *, timeout=10, handle_response=None, count_reply=0 + method, url, data=None, *, timeout=10, handle_response=None, count_reply=0, + max_retries=10, retry_sleep=3, ): + """Send an HTTP request to a will-executor server. + + ``max_retries`` / ``retry_sleep`` control the timeout-retry behaviour: + + * For *critical* operations (pushing inheritance transactions) the + historical default of up to 10 retries with a 3s back-off is kept, so + a transient network hiccup does not lose a transaction. + * For *interactive* operations (ping / info / list download) callers + should pass ``max_retries=0`` so a dead server fails fast (one short + timeout) instead of blocking the UI for minutes. See + :meth:`ping_servers_parallel`. + """ network = Network.get_instance() if not network: raise Exception("You are offline.") @@ -178,9 +191,12 @@ class Willexecutors: else: raise Exception(f"unexpected {method=!r}") except TimeoutError: - if count_reply < 10: - _logger.debug(f"timeout({count_reply}) error: retry in 3 sec...") - time.sleep(3) + if count_reply < max_retries: + _logger.debug( + f"timeout({count_reply}) error: retry in {retry_sleep} sec..." + ) + if retry_sleep: + time.sleep(retry_sleep) return Willexecutors.send_request( method, url, @@ -188,6 +204,8 @@ class Willexecutors: timeout=timeout, handle_response=handle_response, count_reply=count_reply + 1, + max_retries=max_retries, + retry_sleep=retry_sleep, ) else: _logger.debug(f"Too many timeouts: {count_reply}") @@ -254,18 +272,28 @@ class Willexecutors: Willexecutors.get_info_task(url, we) @staticmethod - def get_info_task(url, willexecutor): + def get_info_task(url, willexecutor, *, timeout=DEFAULT_TIMEOUT, + max_retries=0, retry_sleep=0): w = None try: _logger.info("GETINFO_WILLEXECUTOR") _logger.debug(url) - w = Willexecutors.send_request("get", url + "/" + chainname + "/info") + # Fast-fail by default (max_retries=0): a dead server returns after a + # single short timeout instead of retrying 10x with sleeps, which + # used to freeze the UI for minutes per unreachable server. + w = Willexecutors.send_request( + "get", url + "/" + chainname + "/info", + timeout=timeout, max_retries=max_retries, retry_sleep=retry_sleep, + ) if isinstance(w, dict): willexecutor["url"] = url willexecutor["status"] = 200 willexecutor["base_fee"] = w["base_fee"] willexecutor["address"] = w["address"] willexecutor["info"] = w["info"] + else: + # No dict reply (timeout / empty) -> mark as unreachable. + willexecutor["status"] = "KO" _logger.debug(f"response_data {w}") except Exception as e: _logger.error(f"error {e} contacting {url}: {w}") @@ -274,6 +302,109 @@ class Willexecutors: willexecutor["last_update"] = datetime.now().timestamp() return willexecutor + @staticmethod + def ping_servers_parallel(willexecutors, *, on_each=None, max_workers=8, + timeout=DEFAULT_TIMEOUT): + """Ping every will-executor concurrently and report results as they + arrive. + + Network requests run in a thread pool: each ``send_http_on_proxy`` call + schedules its coroutine on Electrum's shared asyncio loop and blocks + only its *own* worker thread, so the total wall-clock time is roughly + that of the slowest server rather than the *sum* of all of them. A + single dead server can no longer stall the whole batch. + + Args: + willexecutors: ``{url: we_dict}`` mapping (mutated in place with the + ping result, exactly like the old sequential ``ping_servers``). + on_each: optional ``callback(url, we_dict, ok: bool)`` invoked from a + worker thread each time a server answers (or fails), so the GUI + can update its list live. Must be thread-safe / marshalled to + the GUI thread by the caller. + max_workers: maximum number of concurrent pings. + timeout: per-request timeout in seconds (fast-fail, no retries). + + Returns: + The same ``willexecutors`` mapping, updated in place. + """ + from concurrent.futures import ThreadPoolExecutor, as_completed + + items = list(willexecutors.items()) + if not items: + return willexecutors + + def _ping_one(url, we): + we = Willexecutors.get_info_task( + url, we, timeout=timeout, max_retries=0, retry_sleep=0 + ) + ok = we.get("status") == 200 + return url, we, ok + + workers = max(1, min(max_workers, len(items))) + with ThreadPoolExecutor(max_workers=workers, + thread_name_prefix="bal-ping") as pool: + futures = [pool.submit(_ping_one, url, we) for url, we in items] + for fut in as_completed(futures): + try: + url, we, ok = fut.result() + except Exception as e: # defensive: never let one server crash all + _logger.error(f"ping_servers_parallel worker error: {e}") + continue + willexecutors[url] = we + if on_each is not None: + try: + on_each(url, we, ok) + except Exception as cb_err: + _logger.error(f"ping on_each callback error: {cb_err}") + return willexecutors + + @staticmethod + def push_transactions_parallel(willexecutors, *, on_each=None, max_workers=8): + """Push transactions to multiple will-executors concurrently. + + Like :meth:`ping_servers_parallel` but for the ``pushtxs`` operation. + Each server keeps the historical retry behaviour of + :meth:`push_transactions_to_willexecutor` (which is important so a real + transaction is not lost to a transient hiccup), but the servers are now + contacted in parallel instead of one-after-another, and results are + reported via ``on_each(url, we_dict, ok, exc)`` as they complete. + + Returns ``{url: (ok, exception_or_None)}``. + """ + from concurrent.futures import ThreadPoolExecutor, as_completed + + targets = [(url, we) for url, we in willexecutors.items() if "txs" in we] + results = {} + if not targets: + return results + + def _push_one(url, we): + try: + ok = Willexecutors.push_transactions_to_willexecutor(we) + return url, we, ok, None + except Willexecutors.AlreadyPresentException as ape: + return url, we, False, ape + except Exception as e: + return url, we, False, e + + workers = max(1, min(max_workers, len(targets))) + with ThreadPoolExecutor(max_workers=workers, + thread_name_prefix="bal-push") as pool: + futures = [pool.submit(_push_one, url, we) for url, we in targets] + for fut in as_completed(futures): + try: + url, we, ok, exc = fut.result() + except Exception as e: + _logger.error(f"push_transactions_parallel worker error: {e}") + continue + results[url] = (ok, exc) + if on_each is not None: + try: + on_each(url, we, ok, exc) + except Exception as cb_err: + _logger.error(f"push on_each callback error: {cb_err}") + return results + @staticmethod def initialize_willexecutor(willexecutor, url, status=None, old_willexecutor=None): old_willexecutor=old_willexecutor if old_willexecutor is not None else {} diff --git a/bal/gui/qt/window.py b/bal/gui/qt/window.py index 5a2005c..cfca79e 100644 --- a/bal/gui/qt/window.py +++ b/bal/gui/qt/window.py @@ -797,48 +797,72 @@ class BalWindow: msg += f"{url}:\t{willexecutors[url]['broadcast_status']}\n" return msg - error = False + # Initialise statuses + show the list immediately. for url in willexecutors: - if self.waiting_dialog._stopping: - return - willexecutor = willexecutors[url] + willexecutors[url].setdefault("broadcast_status", _("waiting...")) + try: self.waiting_dialog.update(getMsg(willexecutors)) - if "txs" in willexecutor: - try: - if Willexecutors.push_transactions_to_willexecutor( - willexecutors[url] - ): - for wid in willexecutors[url]["txsids"]: - self.willitems[wid].set_status("PUSHED", True) - willexecutors[url]["broadcast_status"] = _("Success") - else: - for wid in willexecutors[url]["txsids"]: - self.willitems[wid].set_status("PUSH_FAIL", True) - error = True - willexecutors[url]["broadcast_status"] = _("Failed") - del willexecutor["txs"] - except Willexecutors.AlreadyPresentException: - for wid in willexecutor["txsids"]: - if self.waiting_dialog._stopping: - return - self.waiting_dialog.update( - "checking {} - {} : {}".format( - self.willitems[wid].we["url"], wid, "Waiting" - ) - ) - w = self.willitems[wid] - w.set_check_willexecutor( - Willexecutors.check_transaction(wid, w.we["url"]) - ) - self.waiting_dialog.update( - "checked {} - {} : {}".format( - self.willitems[wid].we["url"], - wid, - self.willitems[wid].get_status("CHECKED"), - ) - ) + except Exception: + pass - if error: + error = {"flag": False} + already_present = [] + + def on_each(url, willexecutor, ok, exc): + # Runs from a worker thread. We only do book-keeping + a thread-safe + # signal-based UI update here; the heavier "already present" check + # path (which itself does network I/O) is handled below in the main + # task thread to keep the original sequential behaviour for it. + if isinstance(exc, Willexecutors.AlreadyPresentException): + already_present.append(url) + willexecutor["broadcast_status"] = _("checking...") + elif ok: + for wid in willexecutor.get("txsids", []): + self.willitems[wid].set_status("PUSHED", True) + willexecutor["broadcast_status"] = _("Success") + else: + for wid in willexecutor.get("txsids", []): + self.willitems[wid].set_status("PUSH_FAIL", True) + error["flag"] = True + willexecutor["broadcast_status"] = _("Failed") + willexecutor.pop("txs", None) + try: + self.waiting_dialog.update(getMsg(willexecutors)) + except Exception: + pass + + if self.waiting_dialog._stopping: + return + # Push to all servers in parallel (each server keeps its own retry + # behaviour, but a slow/dead server no longer blocks the others). + Willexecutors.push_transactions_parallel(willexecutors, on_each=on_each) + + # Handle the "already present" servers: verify each stored tx. This + # keeps the exact original check logic, just executed after the parallel + # push has identified which servers need it. + for url in already_present: + willexecutor = willexecutors[url] + for wid in willexecutor.get("txsids", []): + if self.waiting_dialog._stopping: + return + self.waiting_dialog.update( + "checking {} - {} : {}".format( + self.willitems[wid].we["url"], wid, "Waiting" + ) + ) + w = self.willitems[wid] + w.set_check_willexecutor( + Willexecutors.check_transaction(wid, w.we["url"]) + ) + self.waiting_dialog.update( + "checked {} - {} : {}".format( + self.willitems[wid].we["url"], + wid, + self.willitems[wid].get_status("CHECKED"), + ) + ) + + if error["flag"]: return True def export_json_file(self, path): @@ -941,7 +965,13 @@ class BalWindow: for url in candidates: _logger.info(f"fetch_will_executors_list: trying {url}") try: - resp = Willexecutors.send_request("get", url, timeout=20) + # Fast-fail with a couple of short retries instead of the + # default 10x/3s storm: if the user's connection is flaky we + # want to fall back to the next URL (and then show the simple + # error message) quickly, not freeze for minutes. + resp = Willexecutors.send_request( + "get", url, timeout=10, max_retries=1, retry_sleep=1, + ) _logger.info( f"fetch_will_executors_list: resp type={type(resp).__name__} " f"len={len(resp) if hasattr(resp, '__len__') else 'n/a'}" @@ -997,8 +1027,13 @@ class BalWindow: def ping_willexecutors_task(self, wes): _logger.info("ping willexecutots task") - pinged = [] - failed = [] + # Track per-url state for the live status text. Servers are contacted + # in parallel (see Willexecutors.ping_servers_parallel), so a single + # unreachable server no longer blocks all the others: the whole batch + # now takes about as long as the slowest server instead of the sum of + # every server's (possibly timing-out) request. + pinged = set() + failed = set() def get_title(): msg = _("Ping Will-Executors:") @@ -1006,26 +1041,32 @@ class BalWindow: for url in wes: urlstr = "{:<50}: ".format(url[:50]) if url in pinged: - urlstr += "Ok" + urlstr += _("Ok") elif url in failed: - urlstr += "Ko" + urlstr += _("Ko") else: - urlstr += "--" + urlstr += _("waiting...") urlstr += "\n" msg += urlstr - return msg - for url, we in wes.items(): + def on_each(url, we, ok): + if ok: + pinged.add(url) + else: + failed.add(url) try: self.waiting_dialog.update(get_title()) except Exception: pass - wes[url] = Willexecutors.get_info_task(url, we) - if wes[url]["status"] == "KO": - failed.append(url) - else: - pinged.append(url) + + # Show the initial "waiting..." list immediately. + try: + self.waiting_dialog.update(get_title()) + except Exception: + pass + + Willexecutors.ping_servers_parallel(wes, on_each=on_each) def ping_willexecutors(self, wes, fn_on_success, fn_on_failure=None): def on_success(result): diff --git a/tests/parallel_ping_test.py b/tests/parallel_ping_test.py new file mode 100644 index 0000000..23bbe61 --- /dev/null +++ b/tests/parallel_ping_test.py @@ -0,0 +1,130 @@ +""" +Regression / behaviour test for the parallel will-executor networking. + +Before this change, pinging / pushing to will-executor servers was done in a +sequential loop where every unreachable server blocked the whole batch for the +full timeout (plus up to 10 retries with 3s sleeps). With N servers the total +wall-clock time was the *sum* of every server's time, so a couple of dead +servers froze the GUI ("Non risponde") for minutes. + +This test patches Willexecutors.get_info_task / push_transactions_to_willexecutor +with slow stubs and asserts that: + * ping_servers_parallel contacts servers concurrently (total time ~= the + slowest server, NOT the sum), and + * the on_each callback is invoked once per server with the right ok flag, + * push_transactions_parallel behaves the same way. + +Run with: + QT_QPA_PLATFORM=offscreen PYTHONPATH= \ + python3 tests/parallel_ping_test.py +""" +import importlib +import sys +import time + +PKG = sys.argv[1] if len(sys.argv) > 1 else "electrum.plugins.bal" + +SLOW = 0.5 # seconds each simulated server takes to answer +N = 8 # number of servers + + +def main(): + we_mod = importlib.import_module(f"{PKG}.core.willexecutors") + W = we_mod.Willexecutors + + # ---- 1) ping_servers_parallel: time ~= slowest, not sum ---- + def slow_get_info(url, we, **kwargs): + time.sleep(SLOW) + # half the servers "fail" + if "dead" in url: + we["status"] = "KO" + else: + we["status"] = 200 + return we + + orig_get_info = W.get_info_task + W.get_info_task = staticmethod(slow_get_info) + try: + wes = {} + for i in range(N): + kind = "dead" if i % 2 else "ok" + wes[f"https://{kind}-{i}.example"] = {} + + seen = [] + + def on_each(url, we, ok): + seen.append((url, ok)) + + start = time.time() + W.ping_servers_parallel(wes, on_each=on_each, max_workers=N) + elapsed = time.time() - start + + # Sequential would take ~ N * SLOW. Parallel must be far less. + sequential = N * SLOW + assert elapsed < sequential * 0.6, ( + f"not parallel: {elapsed:.2f}s vs sequential {sequential:.2f}s") + print(f"[OK] ping parallel: {elapsed:.2f}s for {N} servers " + f"(sequential would be ~{sequential:.2f}s)") + + # callback fired once per server, with correct ok flags + assert len(seen) == N, seen + for url, ok in seen: + assert ok == ("ok" in url), (url, ok) + print("[OK] on_each fired once per server with correct ok flag") + + # results written back into the mapping + for url, we in wes.items(): + if "ok" in url: + assert we["status"] == 200, (url, we) + else: + assert we["status"] == "KO", (url, we) + print("[OK] ping results written back into the willexecutors mapping") + finally: + W.get_info_task = orig_get_info + + # ---- 2) push_transactions_parallel: time ~= slowest, not sum ---- + def slow_push(we, **kwargs): + time.sleep(SLOW) + return "fail" not in we["url"] + + orig_push = W.push_transactions_to_willexecutor + W.push_transactions_to_willexecutor = staticmethod(slow_push) + try: + wes = {} + for i in range(N): + kind = "fail" if i % 2 else "good" + wes[f"https://{kind}-{i}.example"] = { + "url": f"https://{kind}-{i}.example", + "txs": "deadbeef", + "txsids": [f"id{i}"], + } + + pushed = [] + + def on_each_push(url, we, ok, exc): + pushed.append((url, ok)) + + start = time.time() + results = W.push_transactions_parallel(wes, on_each=on_each_push, + max_workers=N) + elapsed = time.time() - start + + sequential = N * SLOW + assert elapsed < sequential * 0.6, ( + f"push not parallel: {elapsed:.2f}s vs {sequential:.2f}s") + print(f"[OK] push parallel: {elapsed:.2f}s for {N} servers " + f"(sequential would be ~{sequential:.2f}s)") + + assert len(results) == N, results + for url, (ok, exc) in results.items(): + assert ok == ("good" in url), (url, ok) + print("[OK] push results correct for every server") + finally: + W.push_transactions_to_willexecutor = orig_push + + print(f"\n[OK] parallel networking test passed for package {PKG!r}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_core_heirs.py b/tests/test_core_heirs.py new file mode 100644 index 0000000..f7ea6ab --- /dev/null +++ b/tests/test_core_heirs.py @@ -0,0 +1,266 @@ +""" +Tests for ``bal.core.heirs``. + +Covers constants, OP_RETURN helper, exceptions, validation methods, +and the Heirs model where testable without a live wallet. + +Run: + source electrum/env/bin/activate + python3 tests/test_core_heirs.py +""" + +import sys +import os +sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) + +from bal.core.heirs import ( + HEIR_ADDRESS, HEIR_AMOUNT, HEIR_LOCKTIME, HEIR_REAL_AMOUNT, + HEIR_DUST_AMOUNT, TRANSACTION_LABEL, + create_op_return_script, + AliasNotFoundException, + NotAnAddress, AmountNotValid, LocktimeNotValid, + HeirExpiredException, HeirAmountIsDustException, + NoHeirsException, WillExecutorFeeException, + BalanceTooLowException, + Heirs, +) + + +# ------------------------------------------------------------------ # +# Constants +# ------------------------------------------------------------------ # + +def test_constants(): + assert HEIR_ADDRESS == 0 + assert HEIR_AMOUNT == 1 + assert HEIR_LOCKTIME == 2 + assert HEIR_REAL_AMOUNT == 3 + assert HEIR_DUST_AMOUNT == 4 + assert TRANSACTION_LABEL == "inheritance transaction" + + +# ------------------------------------------------------------------ # +# create_op_return_script +# ------------------------------------------------------------------ # + +def test_op_return_short(): + script = create_op_return_script("42414c") # "BAL" in hex + assert isinstance(script, bytes) + assert script[0] == 0x6a # OP_RETURN + assert len(script) > 3 + + +def test_op_return_long(): + # 76 bytes of data (between 75 and 80) + long_hex = "ab" * 76 + script = create_op_return_script(long_hex) + assert isinstance(script, bytes) + assert script[0] == 0x6a # OP_RETURN + assert script[1] == 0x4c # OP_PUSHDATA1 + + +def test_op_return_empty(): + script = create_op_return_script("") + assert isinstance(script, bytes) + assert len(script) == 2 # OP_RETURN + 0x00 + + +def test_op_return_too_big(): + try: + create_op_return_script("ab" * 81) # 81 bytes > max 80 + assert False, "expected ValueError" + except ValueError: + pass + + +# ------------------------------------------------------------------ # +# Heirs class (without wallet) +# ------------------------------------------------------------------ # + +class FakeDB: + def __init__(self, data=None): + self._data = data or {} + def get(self, key, default=None): + return self._data.get(key, default) + def put(self, key, value): + self._data[key] = value + + +class FakeWallet: + def __init__(self): + self.db = FakeDB({"heirs": { + "alice": ["addr1", "50%", "30d"], + "bob": ["addr2", "10000", "90d"], + }}) + self._dust = 500 + + +def test_heirs_init_from_db(): + wallet = FakeWallet() + heirs = Heirs(wallet) + assert "alice" in heirs + assert "bob" in heirs + assert len(heirs) == 2 + + +def test_heirs_init_empty(): + wallet = FakeWallet() + wallet.db = FakeDB({}) + heirs = Heirs(wallet) + assert len(heirs) == 0 + + +def test_heirs_setitem_saves(): + wallet = FakeWallet() + heirs = Heirs(wallet) + assert len(heirs) == 2 + heirs["charlie"] = ["addr3", "20000", "30d"] + assert "charlie" in heirs + assert "charlie" in wallet.db._data.get("heirs", {}) + + +def test_heirs_pop(): + wallet = FakeWallet() + heirs = Heirs(wallet) + result = heirs.pop("alice") + assert result is not None + assert "alice" not in heirs + assert heirs.pop("nonexistent") is None + + +def test_heirs_check_locktime(): + wallet = FakeWallet() + heirs = Heirs(wallet) + assert heirs.check_locktime() is False + + +def test_heirs_get_locktimes(): + wallet = FakeWallet() + heirs = Heirs(wallet) + # all heirs have locktime "30d" or "90d" -> timestamps > 0 + locktimes = heirs.get_locktimes(0) + assert len(locktimes) >= 1 + for lt in locktimes: + assert lt > 0 + + +def test_heirs_amount_to_float(): + wallet = FakeWallet() + heirs = Heirs(wallet) + + # plain number + assert heirs.amount_to_float(100.5) == 100.5 + # string with percent + assert heirs.amount_to_float("50%") == 50.0 + # invalid -> 0.0 + assert heirs.amount_to_float("notanumber") == 0.0 + + +# ------------------------------------------------------------------ # +# Validation (static methods) +# ------------------------------------------------------------------ # + +def test_validate_address_invalid(): + # This requires a real network, so just verify the exception class + assert issubclass(NotAnAddress, ValueError) + + +def test_validate_amount(): + # Valid percentage + result = Heirs.validate_amount("50%") + assert result == "50%" + + # Valid number + result = Heirs.validate_amount("0.01") + assert result == "0.01" + + # Invalid + try: + Heirs.validate_amount("0.000000001") + assert False, "expected AmountNotValid" + except AmountNotValid: + pass + + try: + Heirs.validate_amount("-1") + assert False, "expected AmountNotValid" + except AmountNotValid: + pass + + +def test_validate_locktime(): + # Valid relative + result = Heirs.validate_locktime("30d") + assert result == "30d" + + result = Heirs.validate_locktime("1y") + assert result == "1y" + + # Empty string returns as-is (no timestamp_to_check, so no validation) + result = Heirs.validate_locktime("") + assert result == "" + + +def test_validate_locktime_expired(): + """A locktime in the past should raise LocktimeNotValid (wrapping HeirExpiredException)""" + import time + past = int(time.time()) - 86400 # yesterday + try: + Heirs.validate_locktime(str(past), timestamp_to_check=past + 1) + assert False, "expected LocktimeNotValid" + except LocktimeNotValid: + pass + + +# ------------------------------------------------------------------ # +# Exceptions +# ------------------------------------------------------------------ # + +def test_alias_not_found(): + exc = AliasNotFoundException() + assert isinstance(exc, Exception) + + +def test_heir_amount_is_dust(): + exc = HeirAmountIsDustException() + assert isinstance(exc, Exception) + + +def test_no_heirs_exception(): + exc = NoHeirsException() + assert isinstance(exc, Exception) + + +def test_will_executor_fee_exception(): + we = {"url": "https://we.example", "base_fee": 1000} + exc = WillExecutorFeeException(we) + assert "WillExecutorFeeException" in str(exc) + assert "1000" in str(exc) + + +def test_balance_too_low_exception(): + exc = BalanceTooLowException(100, 500, 50) + assert "100" in str(exc) + assert "500" in str(exc) + assert "50" in str(exc) + + +# ------------------------------------------------------------------ # +# Heirs static validation (_validate) +# ------------------------------------------------------------------ # + +def test_validate_removes_invalid(): + data = { + "alice": ["addr1", "50%", "30d"], + "bad": ["not_an_address!", "50%", "30d"], + } + result = Heirs._validate(dict(data)) + assert "alice" in result or True # may or may not pass address check + + +if __name__ == "__main__": + for name in sorted(dir()): + if name.startswith("test_"): + globals()[name]() + print(f" [OK] {name}") + print(f"[OK] All heirs tests passed") diff --git a/tests/test_core_heirs_extra.py b/tests/test_core_heirs_extra.py new file mode 100644 index 0000000..79b909a --- /dev/null +++ b/tests/test_core_heirs_extra.py @@ -0,0 +1,182 @@ +""" +Tests for wallet/db-dependent methods in ``bal.core.heirs``. + +Uses mocking to simulate Electrum wallet, db, and bitcoin module. + +Run: + source electrum/env/bin/activate + QT_QPA_PLATFORM=offscreen python3 tests/test_core_heirs_extra.py +""" + +import sys +import os +from unittest.mock import MagicMock, patch + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) + +from bal.core.heirs import ( + Heirs, create_op_return_script, reduce_outputs, + HEIR_ADDRESS, HEIR_AMOUNT, HEIR_LOCKTIME, HEIR_REAL_AMOUNT, +) +from bal.core.willexecutors import Willexecutors + + +# ------------------------------------------------------------------ # +# Heirs db-dependent methods +# ------------------------------------------------------------------ # + +def test_heirs_init_from_db(): + wallet = MagicMock() + wallet.db.get .return_value = {"alice": ["bcrt1q087zm5m3jrhfg78zflqefhcr9heh4c98kzmvhp", 5000, "30d"]} + h = Heirs(wallet) + assert "alice" in h + + +def test_heirs_init_empty_db(): + wallet = MagicMock() + wallet.db.get.return_value = {} + h = Heirs(wallet) + assert len(h) == 0 + + +def test_heirs_save(): + wallet = MagicMock() + wallet.db.get.return_value = {} + h = Heirs(wallet) + h["bob"] = ["bcrt1q08z5t4x74u2883sx2qwsmzk2hj8e5n7z83e4vy", 3000, "60d"] + wallet.db.put.assert_called() + + +def test_heirs_pop_saves(): + wallet = MagicMock() + wallet.db.get.return_value = {"bob": ["bcrt1q08z5t4x74u2883sx2qwsmzk2hj8e5n7z83e4vy", 3000, "60d"]} + h = Heirs(wallet) + h.pop("bob") + wallet.db.put.assert_called() + + +# ------------------------------------------------------------------ # +# Heirs wallet-dependent methods +# ------------------------------------------------------------------ # + +def test_heirs_normalize_perc(): + wallet = MagicMock() + wallet.dust_threshold.return_value = 500 + heir_list = {"a": ["bcrt1q087zm5m3jrhfg78zflqefhcr9heh4c98kzmvhp", "50%", "30d"]} + h = Heirs.__new__(Heirs) + h._Heirs__normal_perc = True + h.update(heir_list) + h.normalize_perc(heir_list, 100000, 100000, wallet) + # "50%" of 100000 = 50000 → above dust threshold, value stays + assert h["a"][HEIR_AMOUNT] == "50%" + + +def test_heirs_prepare_lists(): + wallet = MagicMock() + wallet.dust_threshold.return_value = 500 + h = Heirs.__new__(Heirs) + h.update({"a": ["bcrt1q087zm5m3jrhfg78zflqefhcr9heh4c98kzmvhp", 5000, "30d"]}) + result, onlyfixed = h.prepare_lists(100000, 100, wallet) + assert len(result) > 0 + assert isinstance(result, dict) + + +# ------------------------------------------------------------------ # +# Heirs static methods (pure but use Electrum constants) +# ------------------------------------------------------------------ # + +def test_validate_address_valid(): + with patch("bal.core.heirs.bitcoin.is_address", return_value=True): + result = Heirs.validate_address("1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa") + assert result == "1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa" + + +def test_validate_address_invalid(): + with patch("bal.core.heirs.bitcoin.is_address", return_value=False): + from bal.core.heirs import NotAnAddress + try: + Heirs.validate_address("bad") + assert False, "should have raised" + except NotAnAddress: + pass + + +# ------------------------------------------------------------------ # +# create_op_return_script (pure) +# ------------------------------------------------------------------ # + +def test_create_op_return_script(): + data = "42414c" # "BAL" in hex + script = create_op_return_script(data) + assert script.startswith(b"\x6a") + + +# ------------------------------------------------------------------ # +# reduce_outputs (pure) +# ------------------------------------------------------------------ # + +def test_reduce_outputs_noop(): + outputs = [("bcrt1q087zm5m3jrhfg78zflqefhcr9heh4c98kzmvhp", 1000), ("bcrt1q08z5t4x74u2883sx2qwsmzk2hj8e5n7z83e4vy", 2000)] + reduce_outputs(5000, 5000, 100, outputs) # no crash, no modification + + +def test_reduce_outputs_reduces(): + class FakeOut: + def __init__(self, v): + self.value = v + outputs = [FakeOut(1000), FakeOut(2000)] + reduce_outputs(100, 5000, 10, outputs) + assert outputs[0].value < 1000 + + +# ------------------------------------------------------------------ # +# Willexecutors (pure / light mocking) +# ------------------------------------------------------------------ # + +def test_willexecutors_compute_id(): + wid = Willexecutors.compute_id({"url": "example.com", "chain": "mainnet"}) + assert isinstance(wid, str) + assert "example.com" in wid + + +def test_willexecutors_is_selected(): + assert Willexecutors.is_selected({}) is False + data = {"url": "x"} + assert Willexecutors.is_selected(data) is False + assert Willexecutors.is_selected(data, True) is True + assert data.get("selected") is True + + +def test_willexecutors_get_we_url_from_response(): + class FakeResp: + url = "http://example.com/willexecutor" + result = Willexecutors.get_we_url_from_response(FakeResp()) + # With 4 path segments, result is first 2 segments joined + assert result == "http:/" + # More realistic: a deeper URL returns the host part + class FakeResp2: + url = "http://example.com/api/v1/endpoint" + result2 = Willexecutors.get_we_url_from_response(FakeResp2()) + assert "example.com" in result2 + + +def test_willexecutors_initialize_willexecutor(): + we = {} + Willexecutors.initialize_willexecutor(we, "http://example.com") + assert len(we) > 0 + + +def test_willexecutors_get_willexecutor_transactions_empty(): + assert Willexecutors.get_willexecutor_transactions({}) == {} + + +# ------------------------------------------------------------------ # +# Main +# ------------------------------------------------------------------ # + +if __name__ == "__main__": + for name in sorted(dir()): + if name.startswith("test_"): + globals()[name]() + print(f" [OK] {name}") + print("[OK] All heirs/willexecutors extra tests passed") diff --git a/tests/test_core_plugin_base.py b/tests/test_core_plugin_base.py new file mode 100644 index 0000000..7a83007 --- /dev/null +++ b/tests/test_core_plugin_base.py @@ -0,0 +1,259 @@ +""" +Comprehensive tests for ``bal.core.plugin_base``. + +Covers BalTimestamp, BalConfig, and BalPlugin static helpers. + +Run: + source electrum/env/bin/activate + python3 tests/test_core_plugin_base.py +""" + +import sys +import os +import time +sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) + +from datetime import datetime, date, timedelta +from bal.core.plugin_base import BalTimestamp, BalPlugin, BalConfig + + +# ------------------------------------------------------------------ # +# BalTimestamp +# ------------------------------------------------------------------ # + +def test_bt_create_and_str(): + bt = BalTimestamp("30d") + assert bt.unit == "d" + assert bt.value == 30 + + bt2 = BalTimestamp("1y") + assert bt2.unit == "y" + assert bt2.value == 1 + + bt3 = BalTimestamp(1700000000) + assert bt3.unit is None + assert bt3.value == 1700000000 + + bt4 = BalTimestamp("garbage") + # fallback: value=1, unit=None + assert bt4.value == 1 + assert bt4.unit is None + + bt5 = BalTimestamp(0) + assert bt5.unit is None + assert bt5.value == 0 + + bt6 = BalTimestamp("7d") + assert str(bt6) == "7d" + + bt7 = BalTimestamp("2y") + assert str(bt7) == "2y" + + # absolute timestamp str -> ISO format + bt8 = BalTimestamp(1700000000) + s = str(bt8) + assert "202" in s or "197" in s # year present + + +def test_bt_duration_to_days(): + assert BalTimestamp("30d").duration_to_days() == 30 + assert BalTimestamp("1y").duration_to_days() == 365 + assert BalTimestamp("0d").duration_to_days() == 0 + assert BalTimestamp(1700000000).duration_to_days() == 1700000000 # unit None -> raw value + + +def test_bt_to_date_absolute(): + bt = BalTimestamp(1700000000) + d = bt.to_date() + assert isinstance(d, datetime) + + # absolute with from_date (should be ignored for absolute) + d2 = bt.to_date(from_date=datetime(2020, 1, 1)) + assert d == d2 + + +def test_bt_to_date_relative(): + now = datetime.now() + + # relative days from now + bt = BalTimestamp("7d") + d = bt.to_date() + assert d.hour == 0 and d.minute == 0 # normalized to midnight + assert d > now + + # reverse (subtract days) + d_rev = bt.to_date(reverse=True) + assert d_rev < now + + # from explicit datetime + base = datetime(2025, 6, 1, 12, 0, 0) + d = bt.to_date(from_date=base) + expected = (base + timedelta(days=7)).replace(hour=0, minute=0, second=0, microsecond=0) + assert d == expected + + # from int timestamp + ts = int(base.timestamp()) + d = bt.to_date(from_date=ts) + assert d == expected + + +def test_bt_to_date_years(): + bt = BalTimestamp("1y") + d = bt.to_date() + assert d > datetime.now() + + +def test_bt_to_date_overflow(): + """Huge relative durations should not crash (clamp to INT32_MAX).""" + bt = BalTimestamp("999999999d") + d = bt.to_date() + # should not raise + assert d is not None + assert isinstance(d, datetime) + + +def test_bt_to_timestamp(): + bt = BalTimestamp("7d") + ts = bt.to_timestamp() + assert ts > time.time() + assert isinstance(ts, float) + + bt2 = BalTimestamp(1700000000) + assert abs(bt2.to_timestamp() - 1700000000) < 86400 # close to original + + +def test_bt_repr(): + assert repr(BalTimestamp("7d")) == "7d" + r = repr(BalTimestamp(1700000000)) + assert isinstance(r, str) + assert len(r) > 0 + + +def test_bt_edge_values(): + # zero timestamp + bt0 = BalTimestamp(0) + d = bt0.to_date() + assert d is not None + + # negative? (may depend on platform) + try: + bt_neg = BalTimestamp(-1) + _ = bt_neg.to_date() + except (OSError, ValueError, OverflowError): + pass # acceptable on some platforms + + +# ------------------------------------------------------------------ # +# BalTimestamp._safe_fromtimestamp +# ------------------------------------------------------------------ # + +def test_safe_fromtimestamp_normal(): + d = BalTimestamp._safe_fromtimestamp(1700000000) + assert isinstance(d, datetime) + + +def test_safe_fromtimestamp_nlocktime_max(): + """NLOCKTIME_MAX (2**32-1) must not raise even on 32-bit platforms.""" + d = BalTimestamp._safe_fromtimestamp(2**32 - 1) + assert d is not None + + +def test_safe_fromtimestamp_negative(): + """Negative timestamps should not crash.""" + d = BalTimestamp._safe_fromtimestamp(-1) + assert isinstance(d, datetime) + + +# ------------------------------------------------------------------ # +# BalConfig +# ------------------------------------------------------------------ # + +class FakeConfig: + """Minimal mock for Electrum config.""" + def __init__(self): + self._store = {} + def get(self, key, default=None): + return self._store.get(key, default) + def set_key(self, key, value, save=True): + self._store[key] = value + + +def test_balconfig_default(): + cfg = FakeConfig() + bc = BalConfig(cfg, "test_key", "default_val") + assert bc.get() == "default_val" + assert bc.get("override") == "override" + assert bc.get(None) == "default_val" + + +def test_balconfig_set(): + cfg = FakeConfig() + bc = BalConfig(cfg, "test_key", "default_val") + bc.set("stored_val") + assert cfg.get("test_key") == "stored_val" + assert bc.get() == "stored_val" + + +# ------------------------------------------------------------------ # +# BalPlugin +# ------------------------------------------------------------------ # + +def test_default_will_settings_relative(): + rel = BalPlugin.default_will_settings_relative() + assert rel["threshold"] == "30d" + assert rel["locktime"] == "1y" + + +def test_default_will_settings(): + settings = BalPlugin.default_will_settings() + assert settings["baltx_fees"] == 100 + assert "threshold" in settings + assert "locktime" in settings + # threshold/locktime should be absolute timestamps + assert isinstance(settings["threshold"], float) + assert isinstance(settings["locktime"], float) + + +def test_default_will_settings_absolute(): + abs_ = BalPlugin.default_will_settings_absolute() + assert "threshold" in abs_ + assert "locktime" in abs_ + # should be timestamps (in the future) + today = datetime.combine(date.today(), datetime.min.time()) + assert abs_["threshold"] >= today.timestamp() + assert abs_["locktime"] >= today.timestamp() + + +def test_validate_will_settings(): + # Note: passing None triggers `will_settings = []` which then fails + # on .get(). This is a latent bug — test passing a dict directly + result = BalPlugin.validate_will_settings(None, {"baltx_fees": 0}) + assert result["baltx_fees"] == 100 + + # normal settings unchanged + input_settings = {"baltx_fees": 50, "threshold": 1700000000, "locktime": 1800000000} + result = BalPlugin.validate_will_settings(None, input_settings) + assert result["baltx_fees"] == 50 + assert result["threshold"] == 1700000000 + + +if __name__ == "__main__": + test_bt_create_and_str() + test_bt_duration_to_days() + test_bt_to_date_absolute() + test_bt_to_date_relative() + test_bt_to_date_years() + test_bt_to_date_overflow() + test_bt_to_timestamp() + test_bt_repr() + test_bt_edge_values() + test_safe_fromtimestamp_normal() + test_safe_fromtimestamp_nlocktime_max() + test_safe_fromtimestamp_negative() + test_balconfig_default() + test_balconfig_set() + test_default_will_settings_relative() + test_default_will_settings() + test_default_will_settings_absolute() + test_validate_will_settings() + print(f"[OK] All {sum(1 for k in dir() if k.startswith('test_'))} plugin_base tests passed") diff --git a/tests/test_core_util.py b/tests/test_core_util.py new file mode 100644 index 0000000..3284f75 --- /dev/null +++ b/tests/test_core_util.py @@ -0,0 +1,470 @@ +""" +Comprehensive unit tests for ``bal.core.util.Util``. + +Covers every static method — locktime helpers, amount helpers, comparison +helpers, UTXO helpers, and migration helpers — with edge cases. + +Run: + source electrum/env/bin/activate + python3 tests/test_core_util.py +""" + +import sys +import os +sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) + +from bal.core.util import Util, LOCKTIME_THRESHOLD + + +def test_locktime_to_str(): + # timestamp above threshold -> ISO format + s = Util.locktime_to_str(1700000000) + assert "202" in s and "-" in s, f"expected ISO string, got {s!r}" + + # block height below threshold -> unchanged + assert Util.locktime_to_str(500000) == "500000" + + # string input -> unchanged + assert Util.locktime_to_str("hello") == "hello" + + # zero / edge + assert Util.locktime_to_str(0) == "0" + + +def test_str_to_locktime(): + # relative suffixes pass through + assert Util.str_to_locktime("30d") == "30d" + assert Util.str_to_locktime("1y") == "1y" + assert Util.str_to_locktime("144b") == "144b" + + # integer string -> int + assert isinstance(Util.str_to_locktime("500000"), int) + assert Util.str_to_locktime("500000") == 500000 + + # ISO date -> int timestamp + ts = Util.str_to_locktime("2025-01-01T00:00:00") + assert isinstance(ts, int) + assert ts > 0 + + +def test_parse_locktime_string(): + # plain int -> same int + assert Util.parse_locktime_string(500000) == 500000 + + # int as string -> int + assert Util.parse_locktime_string("500000") == 500000 + + # relative days -> > current timestamp + result = Util.parse_locktime_string("7d") + import time + assert result > time.time() - 86400 + + # relative years -> > current timestamp + result = Util.parse_locktime_string("1y") + assert result > time.time() + + # invalid -> 0 + assert Util.parse_locktime_string("") == 0 + assert Util.parse_locktime_string("garbage") == 0 + + +def test_int_locktime(): + assert Util.int_locktime(seconds=1) == 1 + assert Util.int_locktime(minutes=1) == 60 + assert Util.int_locktime(hours=1) == 3600 + assert Util.int_locktime(days=1) == 86400 + assert Util.int_locktime(blocks=1) == 600 + assert Util.int_locktime(days=1, blocks=1) == 86400 + 600 + assert Util.int_locktime() == 0 + + +def test_encode_decode_amount(): + dp = 8 # typical BTC decimal point + + # percentage passes through + assert Util.encode_amount("50%", dp) == "50%" + assert Util.decode_amount("50%", dp) == "50%" + + # satoshi encoding + assert Util.encode_amount("1.0", dp) == 100000000 + assert Util.encode_amount("0.5", dp) == 50000000 + + # decoding + assert Util.decode_amount(100000000, dp) == "1.00000000" + assert Util.decode_amount(50000000, dp) == "0.50000000" + + # edge + assert Util.encode_amount("abc", dp) == 0 + assert Util.decode_amount("abc", dp) == "abc" + + +def test_is_perc(): + assert Util.is_perc("50%") is True + assert Util.is_perc("100%") is True + assert Util.is_perc("0%") is True + assert Util.is_perc("100") is False + assert Util.is_perc(50) is False + assert Util.is_perc("") is False + assert Util.is_perc(None) is False + + +def test_cmp_array(): + assert Util.cmp_array([1, 2, 3], [1, 2, 3]) is True + assert Util.cmp_array([1, 2, 3], [1, 2]) is False + assert Util.cmp_array([], []) is True + assert Util.cmp_array([1], [2]) is False + assert Util.cmp_array(None, None) is False # exception path + + +def test_cmp_heir(): + heira = ["abc", 10000, 12345] + heirb = ["abc", 10000, 54321] + assert Util.cmp_heir(heira, heirb) is True # addr(0) + amount(1) match + + heirb2 = ["xyz", 10000, 12345] + assert Util.cmp_heir(heira, heirb2) is False # addr mismatch + + heirb3 = ["abc", 20000, 12345] + assert Util.cmp_heir(heira, heirb3) is False # amount mismatch + + +def test_cmp_willexecutor(): + a = {"url": "https://we.example", "address": "bc1abc", "base_fee": 1000} + b = {"url": "https://we.example", "address": "bc1abc", "base_fee": 1000} + assert Util.cmp_willexecutor(a, b) is True + + c = {"url": "https://we.other", "address": "bc1abc", "base_fee": 1000} + assert Util.cmp_willexecutor(a, c) is False + + assert Util.cmp_willexecutor(None, None) is True # None == None + assert Util.cmp_willexecutor({}, {}) is True # both empty + + +def test_search_heir_by_values(): + heirs = { + "alice": {0: "addr1", 1: 1000, 3: 500}, + "bob": {0: "addr2", 1: 2000, 3: 600}, + } + match = Util.search_heir_by_values(heirs, {0: "addr1", 3: 500}, [0, 3]) + assert match == "alice" + + no_match = Util.search_heir_by_values(heirs, {0: "addrX", 3: 500}, [0, 3]) + assert no_match is False + + assert Util.search_heir_by_values({}, {0: "x"}, [0]) is False + + +def test_cmp_heir_by_values(): + a = {0: "addr1", 1: 1000, 3: 500} + b = {0: "addr1", 1: 1000, 3: 500} + assert Util.cmp_heir_by_values(a, b, [0, 1]) is True + assert Util.cmp_heir_by_values(a, b, [0, 1, 3]) is True + + c = {0: "addr1", 1: 9999, 3: 500} + assert Util.cmp_heir_by_values(a, c, [1]) is False + + +def test_cmp_heirs_by_values(): + a = {"h1": {0: "a1", 1: 100}, "h2": {0: "a2", 1: 200}} + b = {"h3": {0: "a1", 1: 100}, "h4": {0: "a2", 1: 200}} + assert Util.cmp_heirs_by_values(a, b, [0, 1]) is True + + c = {"h1": {0: "aX", 1: 100}} + assert Util.cmp_heirs_by_values(a, c, [0, 1]) is False + + +def test_cmp_inputs(): + # Without real TxInput objects we test edge cases + assert Util.cmp_inputs([], []) is True + assert Util.cmp_inputs([1], []) is False + assert Util.cmp_inputs([], [1]) is False + + +def test_cmp_outputs(): + assert Util.cmp_outputs([], []) is True + assert Util.cmp_outputs([1], []) is False + assert Util.cmp_outputs([], [1]) is False + + +def test_cmp_txs(): + # No real Transaction objects, but edge coverage + class FakeTx: + def inputs(self): return [] + def outputs(self): return [] + a = FakeTx() + assert Util.cmp_txs(a, a) is True + + +def test_get_value_amount(): + class FakeOutput: + def __init__(self, addr, val): + self.address = addr + self.value = val + + class FakeTx: + def outputs(self): return self._outs + def __init__(self, outs): self._outs = outs + + # Shared addr+value → both same_amount and same_address → value counted + out_a = FakeOutput("bcrt1q087zm5m3jrhfg78zflqefhcr9heh4c98kzmvhp", 1000) + out_b = FakeOutput("bcrt1q087zm5m3jrhfg78zflqefhcr9heh4c98kzmvhp", 1000) + result = Util.get_value_amount(FakeTx([out_a]), FakeTx([out_b])) + assert result == 1000, f"expected 1000, got {result}" + + # Different address, same amount → same_amount only → not counted + out_c = FakeOutput("bcrt1q08z5t4x74u2883sx2qwsmzk2hj8e5n7z83e4vy", 1000) + result2 = Util.get_value_amount(FakeTx([out_a]), FakeTx([out_c])) + assert result2 == 0, f"expected 0, got {result2}" + + # No matching amount → returns False + out_d = FakeOutput("bcrt1q08z5t4x74u2883sx2qwsmzk2hj8e5n7z83e4vy", 999) + result3 = Util.get_value_amount(FakeTx([out_a]), FakeTx([out_d])) + assert result3 is False, f"expected False, got {result3}" + + +def test_chk_locktime(): + now_ts = 1700000000 + now_block = 800000 + + # timestamp locktime still in future + assert Util.chk_locktime(now_ts, now_block, 1800000000) is True + + # timestamp locktime in past + assert Util.chk_locktime(now_ts, now_block, 1000000000) is False + + # block-height locktime still in future + assert Util.chk_locktime(now_ts, now_block, 900000) is True + + # block-height locktime in past + assert Util.chk_locktime(now_ts, now_block, 100000) is False + + +def test_anticipate_locktime(): + # block-height style (note: "anticipate" actually adds for block locktimes) + result = Util.anticipate_locktime(800000, blocks=100) + assert result == 800000 + 100 + + # timestamp style + ts = 1700000000 + result = Util.anticipate_locktime(ts, days=1) + assert result < ts + assert result > 0 + + # overflow handling (Windows-safe) + huge = 2**32 - 1 # NLOCKTIME_MAX + result = Util.anticipate_locktime(huge, days=1) + assert result > 0 + + # clamp to minimum 1 + low = Util.anticipate_locktime(10, blocks=100) + assert low >= 1 + + +def test_cmp_locktime(): + assert Util.cmp_locktime("30d", "30d") == 0 + # Note: cmp_locktime may return nonzero or None for mismatched units + + +def test_get_locktimes(): + class FakeTx: + locktime = 1700000000 + + # will with single entry + will = { + "tx1": {"tx": FakeTx()}, + } + locktimes = list(Util.get_locktimes(will)) + assert 1700000000 in locktimes + assert len(locktimes) == 1 + + # empty will + assert list(Util.get_locktimes({})) == [] + + +def test_get_lowest_locktimes(): + sorted_ts, sorted_blocks = Util.get_lowest_locktimes([500000, 1700000000, 100, 900000]) + # 500000, 900000 are block-height (< THRESHOLD) + assert 100 in sorted_blocks or True # at least they're sorted + assert 1700000000 in sorted_ts + assert 500000 in sorted_blocks + + # empty + assert Util.get_lowest_locktimes([]) == ([], []) + + +def test_get_will_spent_utxos(): + class FakeTx: + def inputs(self): return [1, 2, 3] + + will = { + "tx1": {"tx": FakeTx()}, + "tx2": {"tx": FakeTx()}, + } + utxos = Util.get_will_spent_utxos(will) + assert len(utxos) == 6 # 3 inputs * 2 txs + + +def test_utxo_to_str(): + class FakeUtxo: + def to_str(self): return "txid:0" + assert Util.utxo_to_str(FakeUtxo()) == "txid:0" + + class FakePrevout: + def to_str(self): return "txid:1" + class FakeUtxo2: + to_str = None + prevout = FakePrevout() + assert Util.utxo_to_str(FakeUtxo2()) == "txid:1" + + # fallback + class Broken: + pass + assert len(Util.utxo_to_str(Broken())) > 0 + + +def test_cmp_utxo(): + class A: + def to_str(self): return "abc:0" + assert Util.cmp_utxo(A(), A()) is True + + class B: + def to_str(self): return "xyz:1" + assert Util.cmp_utxo(A(), B()) is False + + +def test_in_utxo(): + class U: + def __init__(self, s): + self._s = s + def to_str(self): return self._s + + utxos = [U("a:0"), U("b:1")] + target = U("a:0") + assert Util.in_utxo(target, utxos) is True + assert Util.in_utxo(U("z:9"), utxos) is False + assert Util.in_utxo(target, []) is False + + +def test_cmp_output(): + class O: + def __init__(self, addr, val): + self.address = addr + self.value = val + assert Util.cmp_output(O("a", 100), O("a", 100)) is True + assert Util.cmp_output(O("a", 100), O("b", 100)) is False + assert Util.cmp_output(O("a", 100), O("a", 200)) is False + + +def test_in_output(): + class O: + def __init__(self, addr, val): + self.address = addr + self.value = val + outputs = [O("a", 100), O("b", 200)] + assert Util.in_output(O("a", 100), outputs) is True + assert Util.in_output(O("z", 999), outputs) is False + assert Util.in_output(O("a", 100), []) is False + + +def test_din_output(): + class O: + def __init__(self, addr, val): + self.address = addr + self.value = val + + outputs = [O("a", 100), O("b", 200)] + + # same amount AND same address + same_amt, same_addr = Util.din_output(O("a", 100), outputs) + assert same_amt is True and same_addr is True + + # same amount but different address + same_amt, same_addr = Util.din_output(O("c", 100), outputs) + assert same_amt is True and same_addr is False + + # different amount + same_amt, same_addr = Util.din_output(O("z", 999), outputs) + assert same_amt is False and same_addr is False + + +def test_get_current_height(): + # with no network -> 0 + assert Util.get_current_height(None) == 0 + + +def test_copy(): + d = {"a": 1} + Util.copy(d, {"b": 2}) + assert d == {"a": 1, "b": 2} + + # overwrite + Util.copy(d, {"a": 99}) + assert d["a"] == 99 + + +def test_fix_will_settings_tx_fees(): + settings = {"tx_fees": 50} + assert Util.fix_will_settings_tx_fees(settings) is True + assert settings["baltx_fees"] == 50 + assert "tx_fees" not in settings + + # no migration needed + assert Util.fix_will_settings_tx_fees({}) is False + + +def test_fix_will_tx_fees(): + will = { + "tx1": {"tx_fees": 30}, + "tx2": {"baltx_fees": 50}, + } + assert Util.fix_will_tx_fees(will) is True + assert will["tx1"]["baltx_fees"] == 30 + assert "tx_fees" not in will["tx1"] + + # empty will + assert Util.fix_will_tx_fees({}) is False + + +def test_text_hex_conversion(): + assert Util.text_to_hex("BAL") == "42414c" + assert Util.hex_to_text("42414c") == "BAL" + assert Util.text_to_hex("") == "" + assert Util.hex_to_text("") == "" + assert Util.hex_to_text("ZZZ") == "Error: Invalid hex string" + + +if __name__ == "__main__": + test_locktime_to_str() + test_str_to_locktime() + test_parse_locktime_string() + test_int_locktime() + test_encode_decode_amount() + test_is_perc() + test_cmp_array() + test_cmp_heir() + test_cmp_willexecutor() + test_search_heir_by_values() + test_cmp_heir_by_values() + test_cmp_heirs_by_values() + test_cmp_inputs() + test_cmp_outputs() + test_cmp_txs() + test_get_value_amount() + test_chk_locktime() + test_anticipate_locktime() + test_cmp_locktime() + test_get_locktimes() + test_get_lowest_locktimes() + test_get_will_spent_utxos() + test_utxo_to_str() + test_cmp_utxo() + test_in_utxo() + test_cmp_output() + test_in_output() + test_din_output() + test_get_current_height() + test_copy() + test_fix_will_settings_tx_fees() + test_fix_will_tx_fees() + test_text_hex_conversion() + print(f"[OK] All {sum(1 for k in dir() if k.startswith('test_'))} Util tests passed") diff --git a/tests/test_core_will.py b/tests/test_core_will.py new file mode 100644 index 0000000..54a244c --- /dev/null +++ b/tests/test_core_will.py @@ -0,0 +1,272 @@ +""" +Tests for ``bal.core.will``. + +Covers WillItem, Will static methods, and exception classes. + +Run: + source electrum/env/bin/activate + python3 tests/test_core_will.py +""" + +import sys +import os +import copy +sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) + +from bal.core.will import WillItem, Will +from bal.core.willexecutors import Willexecutors + +# A valid serialized Bitcoin transaction hex (1 input + 1 P2PKH output, version 2) +_VALID_TX_HEX = ( + "01000000012a5c9a94fcde98f5581cd00162c60a13936ceb75389ea65b" + "f38633b424eb4031000000006c493046022100a82bbc57a0136751e543" + "3f41cf000b3f1a99c6744775e76ec764fb78c54ee100022100f9e80b7d" + "e89de861dc6fb0c1429d5da72c2b6b2ee2406bc9bfb1beedd729d98501" + "2102e61d176da16edd1d258a200ad9759ef63adf8e14cd97f53227bae3" + "5cdb84d2f6ffffffff0140420f00000000001976a914230ac37834073a" + "42146f11ef8414ae929feaafc388ac00000000" +) + + +def _make_minimal_willitem_dict(**overrides): + """Return a minimal dict that can construct a WillItem.""" + d = { + "tx": _VALID_TX_HEX, + "heirs": {"alice": ["addr1", 5000, "30d"]}, + "willexecutor": None, + "status": "", + "description": "", + "time": 0, + "change": "", + "baltx_fees": 100, + } + d.update(overrides) + return d + + +def _make_willitem_blank(): + """Create a fresh WillItem from scratch.""" + item = WillItem(_make_minimal_willitem_dict()) + # Reset STATUS to clean defaults + item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT) + return item + + +def test_willitem_default_status(): + assert WillItem.STATUS_DEFAULT["VALID"][1] is True + assert WillItem.STATUS_DEFAULT["COMPLETE"][1] is False + assert WillItem.STATUS_DEFAULT["INVALIDATED"][1] is False + assert WillItem.STATUS_DEFAULT["REPLACED"][1] is False + + +def test_willitem_set_get_status(): + # Create a WillItem from a copy of another to avoid tx parsing issues + item = _make_willitem_blank() + assert item.get_status("VALID") is True + + result = item.set_status("COMPLETE", True) + assert result is True + assert item.get_status("COMPLETE") is True + + # Setting to same value returns None + result = item.set_status("COMPLETE", True) + assert result is None + + +def test_willitem_invalidated_clears_valid(): + item = _make_willitem_blank() + assert item.get_status("VALID") is True + + item.set_status("INVALIDATED", True) + assert item.get_status("INVALIDATED") is True + assert item.get_status("VALID") is False # INVALIDATED clears VALID + + +def test_willitem_replaced_clears_valid(): + item = _make_willitem_blank() + item.set_status("REPLACED", True) + assert item.get_status("VALID") is False + + +def test_willitem_pushed_clears_push_fail(): + item = _make_willitem_blank() + item.set_status("PUSH_FAIL", True) + assert item.get_status("PUSH_FAIL") is True + + item.set_status("PUSHED", True) + assert item.get_status("PUSHED") is True + assert item.get_status("PUSH_FAIL") is False + assert item.get_status("CHECK_FAIL") is False + + +def test_willitem_checked_sets_pushed(): + item = _make_willitem_blank() + item.set_status("CHECKED", True) + assert item.get_status("PUSHED") is True + assert item.get_status("PUSH_FAIL") is False + + +def test_willitem_to_dict(): + item = _make_willitem_blank() + d = item.to_dict() + assert "heirs" in d + assert "tx" in d + assert "VALID" in d + + +def test_willitem_str_repr(): + item = _make_willitem_blank() + s = str(item) + assert isinstance(s, str) + r = repr(item) + assert r == s + + +# ------------------------------------------------------------------ # +# Will static methods +# ------------------------------------------------------------------ # + +def test_will_get_sorted_will(): + # Use a simple dict structure that will[key]["tx"].locktime works + class FakeTx: + def __init__(self, locktime): + self.locktime = locktime + + will = { + "b": {"tx": FakeTx(200)}, + "a": {"tx": FakeTx(100)}, + } + sorted_will = Will.get_sorted_will(will) + assert len(sorted_will) == 2 + assert sorted_will[0][1]["tx"].locktime == 100 + assert sorted_will[1][1]["tx"].locktime == 200 + + +def test_will_only_valid(): + item1 = _make_willitem_blank() + item2 = _make_willitem_blank() + item2.set_status("INVALIDATED", True) + + will = {"a": item1, "b": item2} + valid = list(Will.only_valid(will)) + assert "a" in valid + assert "b" not in valid + + +def test_will_only_valid_list(): + item1 = _make_willitem_blank() + item2 = _make_willitem_blank() + item2.set_status("INVALIDATED", True) + + will = {"a": item1, "b": item2} + result = Will.only_valid_list(will) + assert "a" in result + assert "b" not in result + + +def test_will_is_new(): + item1 = _make_willitem_blank() + item1.set_status("COMPLETE", True) + item2 = _make_willitem_blank() # VALID but not COMPLETE + will = {"a": item1, "b": item2} + assert Will.is_new(will) is True + + +def test_will_get_min_locktime(): + class FakeTx: + def __init__(self, locktime): + self.locktime = locktime + + class FakeItem: + def __init__(self, locktime, valid=True): + self.tx = FakeTx(locktime) + self._valid = valid + def get_status(self, s): + return self._valid if s == "VALID" else False + + will = { + "a": FakeItem(100), + "b": FakeItem(200), + } + assert Will.get_min_locktime(will) == 100 + + # empty will + assert Will.get_min_locktime({}) is None + assert Will.get_min_locktime({}, default_value=999) == 999 + + +def test_will_utxos_strs(): + class FakeUtxo: + def __init__(self, s): + self._s = s + def to_str(self): return self._s + + utxos = [FakeUtxo("a:0"), FakeUtxo("b:1")] + strs = Will.utxos_strs(utxos) + assert strs == ["a:0", "b:1"] + assert Will.utxos_strs([]) == [] + + +def test_will_get_tx_from_any(): + tx = Will.get_tx_from_any(_VALID_TX_HEX) + assert tx is not None + assert hasattr(tx, "txid") + + +def test_will_check_tx_height(): + class FakeWallet: + class TxInfo: + tx_mined_status = type("MS", (), {"height": lambda self: 100})() + def get_tx_info(self, tx): + return self.TxInfo() + + wallet = FakeWallet() + assert Will.check_tx_height("fake_tx", wallet) == 100 + + +# ------------------------------------------------------------------ # +# Exception classes +# ------------------------------------------------------------------ # + +def test_exceptions(): + from bal.core.will import ( + WillException, WillExpiredException, NotCompleteWillException, + HeirChangeException, TxFeesChangedException, HeirNotFoundException, + WillexecutorChangeException, NoWillExecutorNotPresent, + WillExecutorNotPresent, NoHeirsException, + AmountException, PercAmountException, FixedAmountException, + ) + + assert issubclass(WillExpiredException, WillException) + assert issubclass(NotCompleteWillException, WillException) + assert issubclass(HeirChangeException, NotCompleteWillException) + assert issubclass(TxFeesChangedException, NotCompleteWillException) + assert issubclass(HeirNotFoundException, NotCompleteWillException) + assert issubclass(WillexecutorChangeException, NotCompleteWillException) + assert issubclass(NoWillExecutorNotPresent, NotCompleteWillException) + assert issubclass(WillExecutorNotPresent, NotCompleteWillException) + assert issubclass(NoHeirsException, WillException) + assert issubclass(PercAmountException, AmountException) + assert issubclass(FixedAmountException, AmountException) + + # WillException default message + exc = WillException() + assert str(exc) == "WillException" + exc2 = WillException("custom") + assert str(exc2) == "custom" + + # WillExpiredException + exc3 = WillExpiredException() + assert isinstance(exc3, WillException) + + +# ------------------------------------------------------------------ # +# Main +# ------------------------------------------------------------------ # + +if __name__ == "__main__": + for name in sorted(dir()): + if name.startswith("test_"): + globals()[name]() + print(f" [OK] {name}") + print(f"[OK] All Will tests passed") diff --git a/tests/test_core_will_extra.py b/tests/test_core_will_extra.py new file mode 100644 index 0000000..0331f9d --- /dev/null +++ b/tests/test_core_will_extra.py @@ -0,0 +1,167 @@ +""" +Tests for wallet-dependent methods in ``bal.core.will``. + +Uses mocking to simulate Electrum wallet, network, and db. + +Run: + source electrum/env/bin/activate + QT_QPA_PLATFORM=offscreen python3 tests/test_core_will_extra.py +""" + +import sys +import os +from unittest.mock import MagicMock, patch, PropertyMock, call + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) + +from bal.core.will import Will, WillItem +from electrum.transaction import Transaction + +_VALID_TX_HEX = ( + "01000000012a5c9a94fcde98f5581cd00162c60a13936ceb75389ea65b" + "f38633b424eb4031000000006c493046022100a82bbc57a0136751e543" + "3f41cf000b3f1a99c6744775e76ec764fb78c54ee100022100f9e80b7d" + "e89de861dc6fb0c1429d5da72c2b6b2ee2406bc9bfb1beedd729d98501" + "2102e61d176da16edd1d258a200ad9759ef63adf8e14cd97f53227bae3" + "5cdb84d2f6ffffffff0140420f00000000001976a914230ac37834073a" + "42146f11ef8414ae929feaafc388ac00000000" +) + + +# Patch Transaction.add_info_from_wallet so it's a no-op during all tests +_patcher = patch.object(Transaction, "add_info_from_wallet") +_patcher.start() + + +# ------------------------------------------------------------------ # +# Will.check_tx_height +# ------------------------------------------------------------------ # + +def test_check_tx_height(): + wallet = MagicMock() + wallet.get_tx_info.return_value.tx_mined_status.height.return_value = 100 + tx = MagicMock() + assert Will.check_tx_height(tx, wallet) == 100 + + +def test_check_tx_height_zero(): + wallet = MagicMock() + wallet.get_tx_info.return_value.tx_mined_status.height.return_value = 0 + tx = MagicMock() + assert Will.check_tx_height(tx, wallet) == 0 + + + + + +# ------------------------------------------------------------------ # +# Will.add_info_from_will +# ------------------------------------------------------------------ # + +def test_add_info_from_will(): + wallet = MagicMock() + willitem = MagicMock() + will = {"wid": willitem} + Will.add_info_from_will(will, "wid", wallet) + willitem.tx.add_info_from_wallet.assert_called_once_with(wallet) + + +def test_add_info_from_will_no_wallet(): + willitem = MagicMock() + will = {"wid": willitem} + Will.add_info_from_will(will, "wid", None) + + +def test_add_info_from_will_tx_is_str(): + wallet = MagicMock() + willitem = MagicMock() + willitem.tx = _VALID_TX_HEX + will = {"wid": willitem} + Will.add_info_from_will(will, "wid", wallet) + assert hasattr(willitem.tx, "add_info_from_wallet") + + +# ------------------------------------------------------------------ # +# Will.check_invalidated +# ------------------------------------------------------------------ # + +def test_check_invalidated_confirmed(): + wallet = MagicMock() + wallet.get_tx_info.return_value.tx_mined_status.height.return_value = 100 + item = WillItem({"tx": _VALID_TX_HEX, "heirs": {"a": ["addr", 100, "30d"]}, + "willexecutor": None, "status": "", "description": "", + "time": 0, "change": "", "baltx_fees": 100}) + will = {"wid": item} + Will.check_invalidated(will, [], wallet) + assert item.get_status("CONFIRMED") is True + + +def test_check_invalidated_pending(): + wallet = MagicMock() + wallet.get_tx_info.return_value.tx_mined_status.height.return_value = 0 + item = WillItem({"tx": _VALID_TX_HEX, "heirs": {"a": ["addr", 100, "30d"]}, + "willexecutor": None, "status": "", "description": "", + "time": 0, "change": "", "baltx_fees": 100}) + will = {"wid": item} + Will.check_invalidated(will, [], wallet) + assert item.get_status("PENDING") is True + + +def test_check_invalidated_invalidated(): + wallet = MagicMock() + wallet.get_tx_info.return_value.tx_mined_status.height.return_value = -1 + item = WillItem({"tx": _VALID_TX_HEX, "heirs": {"a": ["addr", 100, "30d"]}, + "willexecutor": None, "status": "", "description": "", + "time": 0, "change": "", "baltx_fees": 100}) + will = {"wid": item} + Will.check_invalidated(will, [], wallet) + assert item.get_status("INVALIDATED") is True + + +# ------------------------------------------------------------------ # +# Will.check_will (exercises check_invalidated + search_rai) +# ------------------------------------------------------------------ # + +def test_check_will(): + wallet = MagicMock() + wallet.get_tx_info.return_value.tx_mined_status.height.return_value = 0 + item = WillItem({"tx": _VALID_TX_HEX, "heirs": {"a": ["addr", 100, "30d"]}, + "willexecutor": None, "status": "", "description": "", + "time": 0, "change": "", "baltx_fees": 100}) + will = {"wid": item} + Will.check_will(will, [], wallet, 100, 9999999999) + # should be PENDING (height=0) + assert item.get_status("PENDING") is True + + +# ------------------------------------------------------------------ # +# WillItem.__init__ with wallet +# ------------------------------------------------------------------ # + +def test_willitem_init_with_wallet(): + wallet = MagicMock() + w = {"tx": _VALID_TX_HEX, "heirs": {"a": ["addr", 100, "30d"]}, + "willexecutor": None, "status": "", "description": "", + "time": 0, "change": "", "baltx_fees": 100} + item = WillItem(w, wallet=wallet) + assert item is not None + + +def test_willitem_init_without_wallet(): + w = {"tx": _VALID_TX_HEX, "heirs": {"a": ["addr", 100, "30d"]}, + "willexecutor": None, "status": "", "description": "", + "time": 0, "change": "", "baltx_fees": 100} + item = WillItem(w) + assert item is not None + + +# ------------------------------------------------------------------ # +# Main +# ------------------------------------------------------------------ # + +if __name__ == "__main__": + for name in sorted(dir()): + if name.startswith("test_"): + globals()[name]() + print(f" [OK] {name}") + print("[OK] All will-extra tests passed") diff --git a/tests/test_gui_calendar.py b/tests/test_gui_calendar.py new file mode 100644 index 0000000..0f42b82 --- /dev/null +++ b/tests/test_gui_calendar.py @@ -0,0 +1,142 @@ +""" +Tests for ``bal.gui.qt.calendar``. + +Covers BalCalendar static methods: format_time, ical_escape, fold_ical_line, +write_temp_ics, open_with_default_app. + +Run: + QT_QPA_PLATFORM=offscreen python3 tests/test_gui_calendar.py +""" + +import os +import sys +import tempfile +from datetime import datetime, timezone + +sys.path.insert(0, __file__.rsplit("/", 2)[0]) + +from bal.gui.qt.calendar import BalCalendar + + +# ------------------------------------------------------------------ # +# format_time +# ------------------------------------------------------------------ # + +def test_format_time_utc(): + dt = datetime(2025, 6, 1, 12, 30, 45, tzinfo=timezone.utc) + assert BalCalendar.format_time(dt) == "20250601T123045Z" + + +def test_format_time_non_utc(): + from datetime import timedelta + tz = timezone(timedelta(hours=2)) + dt = datetime(2025, 1, 15, 8, 0, 0, tzinfo=tz) + result = BalCalendar.format_time(dt) + assert result.endswith("Z") + assert result == "20250115T060000Z" + + +# ------------------------------------------------------------------ # +# ical_escape +# ------------------------------------------------------------------ # + +def test_ical_escape_no_change(): + text = "hello world" + assert BalCalendar.ical_escape(text) == "hello world" + + +def test_ical_escape_backslash(): + assert BalCalendar.ical_escape("a\\b") == "a\\\\b" + + +def test_ical_escape_semicolon(): + assert BalCalendar.ical_escape("a;b") == "a\\;b" + + +def test_ical_escape_comma(): + assert BalCalendar.ical_escape("a,b") == "a\\,b" + + +def test_ical_escape_multiline(): + text = "line1\r\nline2" + result = BalCalendar.ical_escape(text) + assert "\r\n" in result + assert "line1" in result + assert "line2" in result + + +def test_ical_escape_all(): + text = "\\;," + assert BalCalendar.ical_escape(text) == "\\\\\\;\\," + + +# ------------------------------------------------------------------ # +# fold_ical_line +# ------------------------------------------------------------------ # + +def test_fold_ical_line_short(): + line = "SUMMARY:Test" + assert BalCalendar.fold_ical_line(line) == "SUMMARY:Test" + + +def test_fold_ical_line_long(): + line = "X-LONG:" + "a" * 100 + result = BalCalendar.fold_ical_line(line, limit=75) + parts = result.split("\r\n ") + assert len(parts) > 1 + assert result.startswith("X-LONG:") + + +def test_fold_ical_line_unicode(): + line = "DESCRIPTION:" + "\u20ac" * 40 + result = BalCalendar.fold_ical_line(line, limit=75) + assert "\r\n " in result + assert "\u20ac" in result + + +# ------------------------------------------------------------------ # +# write_temp_ics +# ------------------------------------------------------------------ # + +def test_write_temp_ics(): + content = "BEGIN:VCALENDAR\r\nEND:VCALENDAR\r\n" + path = BalCalendar.write_temp_ics(content) + try: + assert os.path.isfile(path) + with open(path, "rb") as f: + assert f.read() == content.encode("utf-8") + finally: + os.unlink(path) + + +def test_write_temp_ics_empty(): + path = BalCalendar.write_temp_ics("") + try: + assert os.path.isfile(path) + with open(path, "rb") as f: + assert f.read() == b"" + finally: + os.unlink(path) + + +# ------------------------------------------------------------------ # +# open_with_default_app +# ------------------------------------------------------------------ # + +def test_open_with_default_app_not_found(): + result = BalCalendar.open_with_default_app( + "/nonexistent/calendar_app", "/tmp/fake.ics" + ) + assert result is False + + +# ------------------------------------------------------------------ # +# Main +# ------------------------------------------------------------------ # + +if __name__ == "__main__": + for name in sorted(dir()): + if name.startswith("test_"): + globals()[name]() + print(f" [OK] {name}") + print("[OK] All calendar tests passed") diff --git a/tests/test_gui_common.py b/tests/test_gui_common.py new file mode 100644 index 0000000..0d12c53 --- /dev/null +++ b/tests/test_gui_common.py @@ -0,0 +1,102 @@ +""" +Tests for ``bal.gui.qt.common``. + +Covers shown_cv, CheckAliveError, add_widget, log_error, export_meta_gui. + +Run: + QT_QPA_PLATFORM=offscreen python3 tests/test_gui_common.py +""" + +import sys +sys.path.insert(0, __file__.rsplit("/", 2)[0]) + +from PyQt6.QtWidgets import QApplication, QGridLayout, QLabel, QWidget + +# Import the module itself, not via "from .common import *" +import bal.gui.qt.common as C + +_app = QApplication.instance() or QApplication(sys.argv) + + +# ------------------------------------------------------------------ # +# shown_cv +# ------------------------------------------------------------------ # + +def test_shown_cv_default(): + cv = C.shown_cv(True) + assert cv.get() is True + + +def test_shown_cv_set(): + cv = C.shown_cv(True) + cv.set(False) + assert cv.get() is False + + +def test_shown_cv_roundtrip(): + cv = C.shown_cv(False) + assert cv.get() is False + cv.set(True) + assert cv.get() is True + cv.set(True) + assert cv.get() is True + + +# ------------------------------------------------------------------ # +# CheckAliveError +# ------------------------------------------------------------------ # + +def test_check_alive_error_default(): + err = C.CheckAliveError(1000000) + assert err.timestamp_to_check == 1000000 + + +def test_check_alive_error_str(): + err = C.CheckAliveError(1000000) + s = str(err) + assert "Check alive expired" in s + assert "1970" in s + + +def test_check_alive_error_subclass(): + assert issubclass(C.CheckAliveError, Exception) + + +# ------------------------------------------------------------------ # +# add_widget +# ------------------------------------------------------------------ # + +def test_add_widget(): + grid = QGridLayout() + parent = QWidget() + label = QLabel("test") + C.add_widget(grid, "Label", label, 0, "Help text") + assert grid.count() == 3 # label + widget + help button + + +def test_add_widget_multiple_rows(): + grid = QGridLayout() + parent = QWidget() + C.add_widget(grid, "A", QLabel("a"), 0, "help_a") + C.add_widget(grid, "B", QLabel("b"), 1, "help_b") + assert grid.count() == 6 + + +# ------------------------------------------------------------------ # +# log_error +# ------------------------------------------------------------------ # + +def test_log_error_no_window(): + C.log_error((Exception, Exception("test"), None)) + + +# ------------------------------------------------------------------ # +# Main +# ------------------------------------------------------------------ # + +if __name__ == "__main__": + for name in sorted(dir()): + if name.startswith("test_"): + globals()[name]() + print(f" [OK] {name}") + print("[OK] All common tests passed") diff --git a/tests/test_gui_theme.py b/tests/test_gui_theme.py new file mode 100644 index 0000000..7d20de6 --- /dev/null +++ b/tests/test_gui_theme.py @@ -0,0 +1,100 @@ +""" +Tests for ``bal.gui.qt.theme``. + +Covers ``status_color`` priority logic. + +Run: + QT_QPA_PLATFORM=offscreen python3 tests/test_gui_theme.py +""" + +import sys +sys.path.insert(0, __file__.rsplit("/", 2)[0]) + +from bal.gui.qt.theme import status_color + + +# ------------------------------------------------------------------ # +# Helpers +# ------------------------------------------------------------------ # + +class FakeWillItem: + def __init__(self, **status_flags): + self._status = dict(status_flags) + def get_status(self, name): + return self._status.get(name, False) + + +# ------------------------------------------------------------------ # +# Priority-ordered statuses +# ------------------------------------------------------------------ # + +def test_color_invalidated(): + assert status_color(FakeWillItem(INVALIDATED=True)) == "#f87838" + + +def test_color_invalidated_overrides_lower(): + item = FakeWillItem(INVALIDATED=True, PENDING=True, COMPLETE=True) + assert status_color(item) == "#f87838" + + +def test_color_replaced(): + assert status_color(FakeWillItem(REPLACED=True)) == "#ff97e9" + + +def test_color_confirmed(): + assert status_color(FakeWillItem(CONFIRMED=True)) == "#bfbfbf" + + +def test_color_pending(): + assert status_color(FakeWillItem(PENDING=True)) == "#ffce30" + + +# ------------------------------------------------------------------ # +# Branching statuses (CHECK_FAIL / CHECKED / PUSH_FAIL / PUSHED / COMPLETE) +# ------------------------------------------------------------------ # + +def test_color_check_fail_not_checked(): + item = FakeWillItem(CHECK_FAIL=True) + assert status_color(item) == "#e83845" + + +def test_color_check_fail_ignored_if_checked(): + item = FakeWillItem(CHECK_FAIL=True, CHECKED=True) + assert status_color(item) == "#8afa6c" + + +def test_color_checked(): + assert status_color(FakeWillItem(CHECKED=True)) == "#8afa6c" + + +def test_color_push_fail(): + assert status_color(FakeWillItem(PUSH_FAIL=True)) == "#e83845" + + +def test_color_pushed(): + assert status_color(FakeWillItem(PUSHED=True)) == "#73f3c8" + + +def test_color_complete(): + assert status_color(FakeWillItem(COMPLETE=True)) == "#2bc8ed" + + +def test_color_default(): + assert status_color(FakeWillItem()) == "#ffffff" + + +def test_color_check_fail_overrides_push_fail(): + item = FakeWillItem(CHECK_FAIL=True, PUSH_FAIL=True) + assert status_color(item) == "#e83845" + + +# ------------------------------------------------------------------ # +# Main +# ------------------------------------------------------------------ # + +if __name__ == "__main__": + for name in sorted(dir()): + if name.startswith("test_"): + globals()[name]() + print(f" [OK] {name}") + print("[OK] All theme tests passed") diff --git a/tests/test_gui_widgets.py b/tests/test_gui_widgets.py new file mode 100644 index 0000000..8debdf1 --- /dev/null +++ b/tests/test_gui_widgets.py @@ -0,0 +1,255 @@ +""" +Tests for ``bal.gui.qt.widgets``. + +Covers testable widgets without requiring a running Electrum wallet: + - ClickableLabel + - BalLineEdit, BalTextEdit, BalCheckBox + - _LockTimeEditor (static/class methods) + - LockTimeRawEdit (numbify, checkbdy, replace_str) + - PercAmountEdit + +Run: + QT_QPA_PLATFORM=offscreen python3 tests/test_gui_widgets.py +""" + +import sys +sys.path.insert(0, __file__.rsplit("/", 2)[0]) + +from PyQt6.QtCore import QTimer +from PyQt6.QtWidgets import QApplication, QWidget + +from electrum.util import DECIMAL_POINT, decimal_point_to_base_unit_name + +_app = QApplication.instance() or QApplication(sys.argv) + + +# ------------------------------------------------------------------ # +# ClickableLabel +# ------------------------------------------------------------------ # + +def test_clickable_label_creation(): + from bal.gui.qt.widgets import ClickableLabel + lbl = ClickableLabel("test") + assert lbl.text() == "test" + assert hasattr(lbl, "doubleClicked") + + +# ------------------------------------------------------------------ # +# BalLineEdit +# ------------------------------------------------------------------ # + +def test_bal_line_edit(): + from bal.gui.qt.common import shown_cv + from bal.gui.qt.widgets import BalLineEdit + cv = shown_cv("initial") + edit = BalLineEdit(cv) + assert edit.text() == "initial" + cv.set("updated") + assert cv.get() == "updated" + + +# ------------------------------------------------------------------ # +# BalTextEdit +# ------------------------------------------------------------------ # + +def test_bal_text_edit(): + from bal.gui.qt.common import shown_cv + from bal.gui.qt.widgets import BalTextEdit + cv = shown_cv("multi\nline") + edit = BalTextEdit(cv) + assert edit.toPlainText() == "multi\nline" + cv.set("changed") + assert cv.get() == "changed" + + +# ------------------------------------------------------------------ # +# BalCheckBox +# ------------------------------------------------------------------ # + +def test_bal_check_box(): + from bal.gui.qt.common import shown_cv + from bal.gui.qt.widgets import BalCheckBox + cv = shown_cv(True) + cb = BalCheckBox(cv) + assert cb.isChecked() is True + cv.set(False) + assert cv.get() is False + + +def test_bal_check_box_on_click(): + from bal.gui.qt.common import shown_cv + from bal.gui.qt.widgets import BalCheckBox + calls = [] + def handler(): + calls.append(1) + cv = shown_cv(True) + cb = BalCheckBox(cv, on_click=handler) + cb.click() + assert len(calls) == 1 + + +# ------------------------------------------------------------------ # +# _LockTimeEditor +# ------------------------------------------------------------------ # + +def test_locktime_editor_is_acceptable(): + from bal.gui.qt.widgets import _LockTimeEditor + assert _LockTimeEditor.is_acceptable_locktime(100) is True + assert _LockTimeEditor.is_acceptable_locktime(0) is True + assert _LockTimeEditor.is_acceptable_locktime(-1) is False + assert _LockTimeEditor.is_acceptable_locktime(None) is True + + +def test_locktime_editor_is_acceptable_string(): + from bal.gui.qt.widgets import _LockTimeEditor + assert _LockTimeEditor.is_acceptable_locktime("100") is True + assert _LockTimeEditor.is_acceptable_locktime("abc") is False + assert _LockTimeEditor.is_acceptable_locktime("") is True + + +def test_locktime_editor_min_max(): + from bal.gui.qt.widgets import _LockTimeEditor + assert _LockTimeEditor.min_allowed_value >= 0 + assert _LockTimeEditor.max_allowed_value > _LockTimeEditor.min_allowed_value + + +# ------------------------------------------------------------------ # +# LockTimeRawEdit +# ------------------------------------------------------------------ # + +def test_locktime_raw_edit_replace_str(): + from bal.gui.qt.widgets import LockTimeRawEdit + assert LockTimeRawEdit.replace_str("123d") == "123" + assert LockTimeRawEdit.replace_str("456y") == "456" + assert LockTimeRawEdit.replace_str("789b") == "789" + assert LockTimeRawEdit.replace_str("12d34y56b") == "123456" + + +def test_locktime_raw_edit_checkbdy(): + from bal.gui.qt.widgets import LockTimeRawEdit + # character at expected position matches appendix + pos, s = LockTimeRawEdit.checkbdy(None, "123d", 4, "d") + assert s == "123d" + # character at expected position does not match + pos, s = LockTimeRawEdit.checkbdy(None, "123x", 4, "d") + assert s == "123x" + + +def test_locktime_raw_edit_numbify_empty(): + parent = QWidget() + from bal.gui.qt.widgets import LockTimeRawEdit + edit = LockTimeRawEdit(parent) + edit.setText("") + edit.numbify() + assert edit.text() == "" + + +def test_locktime_raw_edit_numbify_days(): + parent = QWidget() + from bal.gui.qt.widgets import LockTimeRawEdit + edit = LockTimeRawEdit(parent) + # Use setText + numbify to simulate user typing + edit.blockSignals(True) + edit.setText("30d") + edit.blockSignals(False) + edit.numbify() + # Should be "30d" with isdays=True + assert edit.text() == "30d" + + +def test_locktime_raw_edit_numbify_years(): + parent = QWidget() + from bal.gui.qt.widgets import LockTimeRawEdit + edit = LockTimeRawEdit(parent) + edit.blockSignals(True) + edit.setText("2y") + edit.blockSignals(False) + edit.numbify() + assert edit.text() == "2y" + + +def test_locktime_raw_edit_get_set_value(): + parent = QWidget() + from bal.gui.qt.widgets import LockTimeRawEdit + edit = LockTimeRawEdit(parent) + edit.set_value("90d") + val = edit.get_value() + assert val is not None + assert "d" in val + + +# ------------------------------------------------------------------ # +# PercAmountEdit +# ------------------------------------------------------------------ # + +def test_perc_amount_edit_numbify_percent(): + from bal.gui.qt.widgets import PercAmountEdit + parent = QWidget() + edit = PercAmountEdit(8, parent=parent) + edit.blockSignals(True) + edit.setText("50%") + edit.blockSignals(False) + edit.numbify() + assert edit.is_perc is True + # After numbify: "50%" -> strip % -> add back -> "50%" + assert edit.text() == "50%" + + +def test_perc_amount_edit_numbify_no_percent(): + from bal.gui.qt.widgets import PercAmountEdit + parent = QWidget() + edit = PercAmountEdit(8, parent=parent) + edit.blockSignals(True) + edit.setText("123") + edit.blockSignals(False) + edit.numbify() + assert edit.is_perc is False + assert edit.text() == "123" + + +def test_perc_amount_get_amount_from_text(): + from bal.gui.qt.widgets import PercAmountEdit + parent = QWidget() + edit = PercAmountEdit(8, parent=parent) + # With percent + result = edit._get_amount_from_text("50%") + assert result is not None + # Without percent + result = edit._get_amount_from_text("123.45") + assert result is not None + # Invalid + result = edit._get_amount_from_text("abc") + assert result is None + + +def test_perc_amount_get_text_from_amount(): + from bal.gui.qt.widgets import PercAmountEdit + parent = QWidget() + edit = PercAmountEdit(lambda: 8, parent=parent) + edit.numbify() # sets is_perc + text = edit._get_text_from_amount(100) + assert isinstance(text, str) + + +def test_perc_amount_get_text_from_amount_perc(): + from bal.gui.qt.widgets import PercAmountEdit + parent = QWidget() + edit = PercAmountEdit(lambda: 8, parent=parent) + edit.blockSignals(True) + edit.setText("50%") + edit.blockSignals(False) + edit.numbify() # sets is_perc = True + text = edit._get_text_from_amount(100) + assert "%" in text + + +# ------------------------------------------------------------------ # +# Main +# ------------------------------------------------------------------ # + +if __name__ == "__main__": + for name in sorted(dir()): + if name.startswith("test_"): + globals()[name]() + print(f" [OK] {name}") + print("[OK] All widget tests passed") diff --git a/tests/test_gui_window_utils.py b/tests/test_gui_window_utils.py new file mode 100644 index 0000000..8a99912 --- /dev/null +++ b/tests/test_gui_window_utils.py @@ -0,0 +1,103 @@ +""" +Tests for ``bal.gui.qt.window_utils``. + +Covers top_level_of, bring_to_front, stop_thread, show_modal, show_on_top. + +Run: + QT_QPA_PLATFORM=offscreen python3 tests/test_gui_window_utils.py +""" + +import sys +sys.path.insert(0, __file__.rsplit("/", 2)[0]) + +from PyQt6.QtCore import QTimer +from PyQt6.QtWidgets import QApplication, QDialog, QWidget + +from bal.gui.qt.window_utils import ( + bring_to_front, show_modal, show_on_top, stop_thread, top_level_of, +) + +_app = QApplication.instance() or QApplication(sys.argv) + + +# ------------------------------------------------------------------ # +# top_level_of +# ------------------------------------------------------------------ # + +def test_top_level_of_child(): + w = QWidget() + child = QWidget(w) + assert top_level_of(child) is w + + +def test_top_level_of_plain_widget(): + w = QWidget() + assert top_level_of(w) is w.window() + + +def test_top_level_of_none(): + assert top_level_of(None) is None + + +def test_top_level_of_dialog(): + d = QDialog() + assert top_level_of(d) is d.window() + + +# ------------------------------------------------------------------ # +# bring_to_front +# ------------------------------------------------------------------ # + +def test_bring_to_front_dialog(): + d = QDialog() + bring_to_front(d) + + +def test_bring_to_front_widget(): + w = QWidget() + bring_to_front(w) + + +# ------------------------------------------------------------------ # +# stop_thread +# ------------------------------------------------------------------ # + +def test_stop_thread_none(): + stop_thread(None) + + +# ------------------------------------------------------------------ # +# show_modal / show_on_top (smoke tests - can't check exec result) +# ------------------------------------------------------------------ # + +def test_show_modal_no_crash(): + d = QDialog() + QTimer.singleShot(0, d.reject) + result = show_modal(d) + assert result == QDialog.DialogCode.Rejected + + +def test_show_on_top_no_crash(): + d = QDialog() + result = show_on_top(d, modal_to_window=True) + assert result is d + d.close() + + +def test_show_on_top_non_modal(): + d = QDialog() + result = show_on_top(d, modal_to_window=False) + assert result is d + d.close() + + +# ------------------------------------------------------------------ # +# Main +# ------------------------------------------------------------------ # + +if __name__ == "__main__": + for name in sorted(dir()): + if name.startswith("test_"): + globals()[name]() + print(f" [OK] {name}") + print("[OK] All window_utils tests passed")