SAFE21.io/bal-electrum-plugin
1
0
Fork 0
forked from bitcoinafterlife/bal-electrum-plugin
Code Pull Requests Activity

feat(net): invio/ping/download verso Will-Executor in parallelo (anti-freeze)

Browse Source 
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.
This commit is contained in:
GenSpark AI Developer
2026-06-15 14:29:04 +00:00
committed by steal
parent b1c8bba9e9
commit 03985a2566
16 changed files with 2843 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

164
REPORT_NETWORKING_PARALLELO.md Normal file
View File

@@ -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=<electrum-src> \
python3 -m pytest tests/ -q
QT_QPA_PLATFORM=offscreen PYTHONPATH=<electrum-src> \
python3 tests/smoke_test.py electrum.plugins.bal
QT_QPA_PLATFORM=offscreen PYTHONPATH=<electrum-src> \
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à).

2
bal/core/util.py
View File

@@ -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:

143
bal/core/willexecutors.py
View File

@@ -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 {}

145
bal/gui/qt/window.py
View File

@@ -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):

130
tests/parallel_ping_test.py Normal file
View File

@@ -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=<electrum-src> \
python3 tests/parallel_ping_test.py <PLUGIN_IMPORT_NAME>
"""
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())

266
tests/test_core_heirs.py Normal file
View File

@@ -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")

182
tests/test_core_heirs_extra.py Normal file
View File

@@ -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")

259
tests/test_core_plugin_base.py Normal file
View File

@@ -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")

470
tests/test_core_util.py Normal file
View File

@@ -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")

272
tests/test_core_will.py Normal file
View File

@@ -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")

167
tests/test_core_will_extra.py Normal file
View File

@@ -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")

142
tests/test_gui_calendar.py Normal file
View File

@@ -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")

102
tests/test_gui_common.py Normal file
View File

@@ -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")

100
tests/test_gui_theme.py Normal file
View File

@@ -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")

255
tests/test_gui_widgets.py Normal file
View File

@@ -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")

103
tests/test_gui_window_utils.py Normal file
View File

@@ -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")