bitcoinafterlife/bal-electrum-plugin
3
2
Fork 4
Code Issues Pull Requests Actions Packages Projects Releases 8 Wiki Activity

gui: rebuild the will automatically on new transactions (AUTO_REBUILD, default off; anticipate delivery by 1 day, invalidate on-chain only when the anticipated locktime crosses the check-alive threshold)

Browse Source
This commit is contained in:
svatantrya
2026-08-16 06:40:35 -04:00
parent 4bbdf261e3
commit e9142a7c1f
5 changed files with 964 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
bal/core/plugin_base.py
View File

@@ -259,6 +259,16 @@ class BalPlugin(BasePlugin):
# invalidation prompts at close. Default ON.
self.REBUILD_ON_CLOSE = BalConfig(config, "bal_rebuild_on_close", True)
# AUTO_REBUILD: when enabled, an incoming/outgoing wallet transaction
# automatically re-runs the same rebuild flow the wizard runs at
# wallet close (anticipate the delivery date by one day to orphan the
# previous will; build an on-chain invalidation tx ONLY when the
# anticipated locktime would fall before the check-alive threshold or
# the threshold is already in the past). When disabled (default) the
# will is only rebuilt when the user presses Check / Prepare or closes
# the wallet. Default OFF.
self.AUTO_REBUILD = BalConfig(config, "bal_auto_rebuild", False)
# EDITABLE_DATES (Group C / C2): when enabled, the delivery-time and
# check-alive date fields are editable everywhere (toolbar / Heirs tab),
# not only inside the "Build your will" wizard. Default OFF, so the dates

81
bal/gui/qt/plugin.py
View File

@@ -15,6 +15,7 @@ and cached in ``self.bal_windows``.
"""
from electrum.gui.qt.main_window import StatusBarButton
from electrum.util import EventListener, event_listener
from PyQt6.QtWidgets import QLayout
from .common import *
@@ -40,7 +41,7 @@ def _window_key(window):
return id(window)
class Plugin(BalPlugin):
class Plugin(BalPlugin, EventListener):
def __init__(self, parent, config, name):
_logger.info("INIT BALPLUGIN")
BalPlugin.__init__(self, parent, config, name)
@@ -49,6 +50,10 @@ class Plugin(BalPlugin):
# remove a stale button before creating a fresh one when a wallet is
# switched / Electrum is restarted, so the icon is never duplicated.
self._statusbar_buttons = {}
# Register the on_event_* handlers with Electrum's callback manager so
# the plugin learns about new wallet transactions (used by the
# AUTO_REBUILD setting).
self.register_callbacks()
@hook
def init_qt(self, gui_object):
@@ -328,6 +333,44 @@ class Plugin(BalPlugin):
except Exception as e:
_logger.error("close_wallet: on_close failed: {}".format(e))
@event_listener
def on_event_new_transaction(self, wallet, tx):
"""Electrum event: a transaction was added to *wallet*."""
self._wallet_activity(wallet)
@event_listener
def on_event_wallet_updated(self, wallet):
"""Electrum event: *wallet* finished a sync pass."""
self._wallet_activity(wallet)
def _wallet_activity(self, wallet):
"""React to wallet activity (new transaction / sync update).
When the AUTO_REBUILD setting is enabled, any change to a wallet that
has a live BalWindow schedules the headless "auto rebuild" flow
(``BalWindow.schedule_auto_rebuild``): it re-runs the same check the
wizard runs at wallet close, anticipating the delivery date by one day
and building an on-chain invalidation tx only when the anticipated
locktime would fall before the check-alive threshold (or the threshold
is already in the past).
This handler runs on the asyncio callback thread, so it only touches
thread-safe state and defers all work to the BalWindow (which marshals
itself onto the GUI thread through QTimer).
"""
if not self.AUTO_REBUILD.get():
return
for win in list(self.bal_windows.values()):
try:
if (
getattr(win, "wallet", None) == wallet
and win.ok
and not win.disable_plugin
):
win.schedule_auto_rebuild()
except Exception as e:
_logger.debug("_wallet_activity failed: {}".format(e))
@hook
def init_keystore(self):
_logger.debug("init keystore")
@@ -464,6 +507,17 @@ class Plugin(BalPlugin):
# ADVANCED).
heir_rebuild_on_close = BalCheckBox(self.REBUILD_ON_CLOSE)
# "Rebuild automatically on new transactions" checkbox. Bound to the
# persisted AUTO_REBUILD config (default OFF). When ticked, an incoming
# or outgoing wallet transaction automatically re-runs the same rebuild
# flow the wizard runs at wallet close: the delivery date is
# anticipated by one day (so the new will replaces the previous one
# without an invalidation tx), and an on-chain invalidation is only
# built when the anticipated locktime would fall before the Check Alive
# threshold or the threshold is already in the past. Visible to all
# users (BASIC and ADVANCED).
heir_auto_rebuild = BalCheckBox(self.AUTO_REBUILD)
# USER TYPE selector (SIMPLE / ADVANCED, global). A two-choice combo
# (not a free-text field) bound to the USER_TYPE config:
# index 0 -> "BASIC" -> stored value "basic" (DEFAULT)
@@ -836,6 +890,30 @@ class Plugin(BalPlugin):
)
grid.addWidget(reset_btn_rebuild_on_close, 15, 3)
# "Rebuild automatically on new transactions" row (always visible,
# BASIC + ADVANCED), right below the "Rebuild will on wallet close"
# row.
lbl_auto_rebuild = QLabel(_("Rebuild automatically on new transactions"))
help_auto_rebuild = HelpButton(
"When a new transaction arrives for the wallet, automatically "
"rebuild the will the same way the wizard does at wallet close: "
"the delivery date is anticipated by one day so the new will "
"replaces the previous one, and the rebuilt transactions are "
"signed and sent to their will-executors.\n"
"An on-chain invalidation transaction is only built when the "
"anticipated delivery date would fall before the Check Alive "
"threshold, or when the threshold is already in the past.\n"
"When disabled (default), the will is only rebuilt on Check / "
"Prepare / wallet close."
)
grid.addWidget(lbl_auto_rebuild, 16, 0)
grid.addWidget(heir_auto_rebuild, 16, 1)
grid.addWidget(help_auto_rebuild, 16, 2)
reset_btn_auto_rebuild = _make_reset_btn(
self.AUTO_REBUILD, heir_auto_rebuild, "check"
)
grid.addWidget(reset_btn_auto_rebuild, 16, 3)
# ----------------------------------------------------------------- #
# Group C / C4b: "Reset" button that restores the dialog settings to #
# their factory defaults. It only resets the settings exposed by THIS #
@@ -869,6 +947,7 @@ class Plugin(BalPlugin):
(self.SAVE_HISTORY, heir_save_history, "check"),
(self.HISTORY_LABEL, edit_history_label, "line"),
(self.REBUILD_ON_CLOSE, heir_rebuild_on_close, "check"),
(self.AUTO_REBUILD, heir_auto_rebuild, "check"),
]
for cfg, widget, kind in resets:
# Persist the default value back into the Electrum config.

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

@@ -39,6 +39,13 @@ from .widgets import LockTimeWidget, PercAmountEdit
class BalWindow:
# Automatic rebuild-on-new-transaction flow (AUTO_REBUILD setting):
# the debounce window collapses bursts of wallet events into one run, and
# the cooldown prevents the flow from re-triggering right after a rebuild
# (the freshly persisted txs can themselves fire wallet events).
_AUTO_REBUILD_DEBOUNCE_MS = 5000
_AUTO_REBUILD_COOLDOWN = 10.0
def __init__(self, bal_plugin: "BalPlugin", window: "ElectrumWindow"):
self.bal_plugin = bal_plugin
self.window = window
@@ -56,6 +63,9 @@ class BalWindow:
# ``init_menubar_tools`` twice would add the Heirs/Will tabs and the
# menu actions twice, producing the garbled/condensed menu entry.
self._menubar_initialized = False
# Auto-rebuild flow state: re-entrancy guard and cooldown deadline.
self._auto_rebuild_running = False
self._auto_rebuild_cooldown_until = 0.0
self.bal_plugin.get_decimal_point = self.window.get_decimal_point
if self.window.wallet:
@@ -1034,6 +1044,297 @@ class BalWindow:
password = self.get_wallet_password(message)
return password
# ------------------------------------------------------------------ #
# Automatic rebuild on new transactions (AUTO_REBUILD)
#
# When the AUTO_REBUILD setting is enabled, wallet activity (a new
# transaction / a sync update) schedules the headless rebuild flow below,
# which reproduces EXACTLY what the "Build your will" wizard does at wallet
# close (task_phase1 / task_phase2):
#
# * the delivery date of the rebuilt transactions is anticipated by one
# day (Will.search_anticipate -> check_anticipate) so the new will
# mines BEFORE the previous one and orphans it WITHOUT an on-chain
# invalidation transaction;
# * an on-chain invalidation transaction is built ONLY when the
# anticipated locktime would fall before the check-alive threshold
# (post-build check_will -> WillExpiredException), or when the
# threshold is already in the past (CheckAliveError) - the same two
# conditions that trigger invalidation in the wizard.
# ------------------------------------------------------------------ #
def schedule_auto_rebuild(self, delay_ms=None):
"""Debounced entry point for the auto-rebuild flow.
Called by ``Plugin._wallet_activity`` (on the asyncio callback thread)
whenever a transaction/update is seen for this wallet. The actual
rebuild is deferred through ``QTimer`` (thread-safe to schedule, runs
on the GUI thread) so a burst of events collapses into a single run.
"""
try:
delay = delay_ms if delay_ms is not None else self._AUTO_REBUILD_DEBOUNCE_MS
QTimer.singleShot(delay, self._run_auto_rebuild)
except Exception as e:
_logger.debug("schedule_auto_rebuild failed: {}".format(e))
def _run_auto_rebuild(self):
"""GUI-thread guard before launching the auto-rebuild worker.
Checks the cheap guards that must be evaluated on the GUI thread and,
when allowed, runs the headless flow in a background thread so the
interface is not frozen (signing/pushing can take a while).
"""
if not self._auto_rebuild_allowed():
return
self._auto_rebuild_running = True
threading.Thread(target=self._auto_rebuild_worker, daemon=True).start()
def _auto_rebuild_worker(self):
try:
self._auto_rebuild_flow()
except Exception as e:
_logger.error("auto rebuild worker failed: {}".format(e))
finally:
self._auto_rebuild_running = False
QTimer.singleShot(0, self._after_auto_rebuild)
def _auto_rebuild_allowed(self):
"""Cheap guards evaluated before running the auto-rebuild flow."""
if self.disable_plugin or not self.ok:
return False
if not self.bal_plugin.AUTO_REBUILD.get():
return False
if not self.willitems:
return False
if self._auto_rebuild_running:
return False
if time.time() < self._auto_rebuild_cooldown_until:
return False
return True
def maybe_auto_rebuild(self):
"""Run the headless auto-rebuild flow synchronously on this thread.
This is the testable entry point (and what the background worker
runs): it reproduces the wizard's close-time flow and returns True when
it rebuilt/invalidated the will, False when there was nothing to do.
"""
if not self._auto_rebuild_allowed():
return False
self._auto_rebuild_running = True
try:
result = self._auto_rebuild_flow()
finally:
self._auto_rebuild_running = False
QTimer.singleShot(0, self._after_auto_rebuild)
return result
def _after_auto_rebuild(self):
"""Refresh the will tabs after an auto-rebuild (GUI thread)."""
try:
self.update_all()
except Exception as e:
_logger.debug("_after_auto_rebuild update_all failed: {}".format(e))
try:
if hasattr(self, "will_list_widget"):
self.will_list_widget.update()
except Exception:
pass
def _auto_rebuild_flow(self):
"""Core headless rebuild flow (mirrors the wizard's close flow).
Returns True when the will was rebuilt or invalidated, False when there
was nothing to do. Runs on the caller's thread.
"""
try:
self._auto_rebuild_cooldown_until = (
time.time() + self._AUTO_REBUILD_COOLDOWN
)
_logger.info("auto rebuild: checking will after wallet activity")
# 1) Recompute date_to_check / willexecutors exactly like
# init_class_variables does at the start of the wizard's phase 1.
# A Check Alive threshold already in the past (ADVANCED mode)
# means the old will must be invalidated on-chain.
try:
self.init_class_variables()
except CheckAliveError:
_logger.info("auto rebuild: check-alive threshold passed -> invalidate")
self._auto_invalidate_will()
return True
except NoHeirsException:
_logger.info("auto rebuild: no heirs, nothing to rebuild")
return False
# 2) Check the current will against the freshly computed reference
# date. A still-valid will needs no rebuild.
try:
self.check_will()
_logger.debug("auto rebuild: will is still valid, nothing to do")
return False
except (WillExpiredException, WillPostponedException) as e:
# Expired ("too late to anticipate") or a postpone on a
# signed/sent will: the old coins must be invalidated on-chain
# first.
_logger.info(
"auto rebuild: {} -> invalidate".format(type(e).__name__)
)
self._auto_invalidate_will()
return True
except NoHeirsException:
return False
except NotCompleteWillException:
# The will no longer covers the wallet's current UTXOs / heirs
# / date: rebuild it. The rebuild automatically anticipates
# the delivery date by one day when the same coins/heirs are
# involved (Will.search_anticipate), so the new transactions
# mine before the previous ones.
pass
# 3) Rebuild.
try:
txs = self.build_will()
except Exception as e:
_logger.error("auto rebuild: build_will failed: {}".format(e))
return False
if not txs:
_logger.info("auto rebuild: nothing was built")
return False
# 4) Re-validate the freshly built will (mirrors task_phase1 after
# build_will). If the anticipated locktime now falls before the
# check-alive threshold, the previous will must be invalidated
# on-chain before the new one is used - and we STOP, exactly like
# the wizard ("invalidate_classic"): signing/pushing the new will
# while the invalidation is not confirmed would race it for the
# same inputs. The next wallet event / manual Check continues
# once the invalidation confirms.
try:
self.check_will()
except (WillExpiredException, WillPostponedException) as e:
_logger.info(
"auto rebuild: anticipated locktime crossed threshold "
"({}) -> invalidate old will".format(type(e).__name__)
)
self._auto_invalidate_will()
return True
except NoHeirsException:
return False
except NotCompleteWillException:
# The freshly rebuilt transactions simply need signing.
pass
except Exception as e:
_logger.error(
"auto rebuild: post-build check failed: {}".format(e)
)
return False
# 5) Sign (passwordless wallets only, headlessly), persist and push
# the rebuilt transactions to their will-executors: pushing the
# earlier-locktime transactions is what makes them orphan the
# previous ones.
self._auto_sign_save_push()
return True
finally:
# Always apply the cooldown so a burst of events (or the wallet
# events fired by our own persistence) cannot loop forever.
self._auto_rebuild_cooldown_until = (
time.time() + self._AUTO_REBUILD_COOLDOWN
)
def _auto_invalidate_will(self, will=None):
"""Build, sign and broadcast the on-chain invalidation tx, headlessly.
Reuses the exact recipe of the wizard's ``loop_broadcast_invalidating``
(label set before broadcast, tx info pulled from wallet/network,
broadcast timeout 120s) without any dialog. An encrypted wallet cannot
sign headlessly, so we stop with a logged warning and leave the
invalidation to the user's manual flow.
"""
willitems = will if will is not None else self.willitems
try:
tx = Will.invalidate_will(
willitems,
self.wallet,
self.will_settings.get("baltx_fees", 1),
history_label=self.bal_plugin.HISTORY_LABEL.get(),
will_locktime=Will.get_min_locktime(
willitems,
default_value=getattr(self, "date_to_check", None),
),
)
except Exception as e:
_logger.error("auto invalidate: could not build tx: {}".format(e))
return None
if not tx:
_logger.info("auto invalidate: no transactions to invalidate")
return None
try:
if self.wallet.has_keystore_encryption():
_logger.warning(
"auto invalidate: wallet is encrypted; signing the "
"invalidation requires the password -> invalidate manually"
)
return None
network = getattr(self.wallet, "network", None)
if network is None:
_logger.error("auto invalidate: no network, cannot broadcast")
return None
tx = self.wallet.sign_transaction(tx, None, ignore_warnings=True)
if not tx or not tx.is_complete():
raise Exception("invalidation tx not complete")
tx.add_info_from_wallet(self.wallet)
network.run_from_another_thread(tx.add_info_from_network(network))
txid = tx.txid()
if txid:
# Label BEFORE broadcasting so the History tab shows it the
# moment the tx appears (matches the wizard behaviour).
self.wallet.set_label(txid, "BAL Invalidate transaction")
network.run_from_another_thread(
network.broadcast_transaction(tx, timeout=120), timeout=120
)
_logger.info("auto invalidate: broadcast invalidation {}".format(txid))
return tx
except Exception as e:
_logger.error("auto invalidate failed: {}".format(e))
return None
def _auto_sign_save_push(self):
"""Headless sign + persist + push of the rebuilt will.
Mirrors the wizard's phase 2 (sign_transactions -> save_willitems ->
push_transactions_to_willexecutors) without dialogs. Encrypted
wallets cannot be signed headlessly, so the rebuilt transactions are
left unsigned ("New") for the user to sign manually.
"""
try:
if self.wallet.has_keystore_encryption():
_logger.warning(
"auto rebuild: wallet is encrypted; rebuilt will left "
"unsigned (sign manually)"
)
else:
txs = self.sign_transactions(None)
if txs:
for txid, tx in txs.items():
# Store the signed tx back, like
# ask_password_and_sign_transactions.on_success does
# (re-parse instead of deepcopy: the signed tx may carry
# wallet-derived input info holding a threading.RLock).
self.willitems[txid].tx = Will.get_tx_from_any(str(tx))
except Exception as e:
_logger.error("auto rebuild: signing failed: {}".format(e))
try:
self.save_willitems()
except Exception as e:
_logger.error("auto rebuild: save_willitems failed: {}".format(e))
self._save_will_to_history()
try:
self.push_transactions_to_willexecutors()
except Exception as e:
_logger.error("auto rebuild: push failed: {}".format(e))
def on_close(self):
# Wallet is closing: run the closing "build will" task and tear down
# the plugin's tabs/menu. Each step is isolated so that one failure

570
tests/test_auto_rebuild_on_new_tx.py Normal file
View File

@@ -0,0 +1,570 @@
#!/usr/bin/env python3
"""Tests for the "Rebuild automatically on new transactions" (AUTO_REBUILD)
feature.
Covers:
* the persisted ``bal_auto_rebuild`` configuration key exists and defaults
to OFF (False), and can be enabled and read back;
* the event wiring: ``Plugin._wallet_activity`` schedules the rebuild only
for the matching wallet and only when the setting is enabled;
* ``BalWindow.schedule_auto_rebuild`` debounces through ``QTimer`` and the
re-entrancy / cooldown guards;
* ``BalWindow.maybe_auto_rebuild`` reproduces the wizard's close-time flow:
- no-op when the will is still valid;
- rebuild + sign + push when a new UTXO invalidates the will (no on-chain
invalidation, the rebuilt tx is anticipated to mine before the old);
- on-chain invalidation when the check-alive threshold is already in the
past (CheckAliveError);
- on-chain invalidation when the will is already expired;
- on-chain invalidation when the anticipated locktime would fall before
the check-alive threshold (and no sign/push in that case).
Run:
source /home/steal/devel/bal/electrum/env/bin/activate
QT_QPA_PLATFORM=offscreen python3 tests/test_auto_rebuild_on_new_tx.py
"""
import os
import sys
import tempfile
import time
import unittest.mock as mock
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from electrum import bitcoin, crypto # noqa: E402
from electrum.descriptor import parse_descriptor # noqa: E402
from electrum.transaction import ( # noqa: E402
PartialTxInput,
PartialTxOutput,
TxOutpoint,
)
from electrum.util import bfh # noqa: E402
from PyQt6.QtWidgets import QApplication # noqa: E402
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import bal.gui.qt.window as window_mod # noqa: E402
from bal.core.heirs import Heirs # noqa: E402
from bal.core.plugin_base import BalConfig, BalPlugin # noqa: E402
from bal.core.util import Util # noqa: E402
from bal.core.will import Will # noqa: E402
from bal.core.willexecutors import Willexecutors # noqa: E402
from bal.gui.qt.plugin import Plugin # noqa: E402
from bal.gui.qt.window import BalWindow # noqa: E402
CONFIG_KEY = "bal_auto_rebuild"
# --------------------------------------------------------------------------- #
# Fixtures
# --------------------------------------------------------------------------- #
PRIVKEY = bytes(range(32))
PUBKEY = crypto.privkey_to_pubkey(PRIVKEY)
ADDRESS = bitcoin.public_key_to_p2wpkh(PUBKEY)
SCRIPT = bitcoin.address_to_script(ADDRESS)
FUNDING_SATOSHIS = 500000
def make_funding_input(prevout_hex="11" * 32):
"""Return a fake wallet UTXO spendable by the will."""
utxo = PartialTxInput(prevout=TxOutpoint(bfh(prevout_hex), 0))
utxo.witness_utxo = PartialTxOutput.from_address_and_value(
ADDRESS, FUNDING_SATOSHIS
)
utxo._trusted_value_sats = FUNDING_SATOSHIS
utxo._TxInput__scriptpubkey = SCRIPT
utxo._TxInput__address = ADDRESS
return utxo
class FakeDB:
def __init__(self):
self._data = {}
def get(self, key, default=None):
return self._data.get(key, default)
def put(self, key, value):
self._data[key] = value
def get_transaction(self, txid):
return None
def commit(self):
pass
class FakeWallet:
def __init__(self, utxos):
self.db = FakeDB()
self.adb = None
self.network = None
self._utxos = list(utxos)
self._dust = 546
self._change_addresses = [ADDRESS]
self.labels = {}
self.save_db_calls = 0
def save_db(self):
self.save_db_calls += 1
def dust_threshold(self):
return self._dust
def has_keystore_encryption(self):
return False
def set_label(self, txid, label):
self.labels[txid] = label
def get_all_labels(self):
return dict(self.labels)
def get_label_for_txid(self, txid):
return self.labels.get(txid, "")
def get_utxos(self):
return list(self._utxos)
def get_change_addresses_for_new_transaction(self, *args, **kwargs):
return self._change_addresses
def add_input_info(self, txin, only_der_suffix=False):
pass
def add_output_info(self, txout, only_der_suffix=False):
pass
def get_tx_info(self, tx):
class _TxInfo:
def __init__(self):
class _MinedStatus:
def height(self):
return 0
self.tx_mined_status = _MinedStatus()
return _TxInfo()
def get_transaction(self, txid):
return None
def sign_transaction(self, tx, password=None, ignore_warnings=True):
descriptor = parse_descriptor(f"wpkh({PUBKEY.hex()})")
for txin in tx.inputs():
if txin.script_descriptor is None:
txin.script_descriptor = descriptor
if txin.value_sats() is None:
txin._trusted_value_sats = FUNDING_SATOSHIS
tx.sign({PUBKEY: PRIVKEY})
class FakeConfig:
def __init__(self):
self._data = {}
self._tmpdir = tempfile.mkdtemp(prefix="bal-test-")
def electrum_path(self):
return self._tmpdir
def user_dir(self):
return self._tmpdir
def get(self, key, default=None):
return self._data.get(key, default)
def set_key(self, key, value, save=True):
self._data[key] = value
class FakeWindow:
def __init__(self, wallet):
self.wallet = wallet
self.messages = []
self.warnings = []
self.errors = []
def get_decimal_point(self):
return 0
def show_message(self, text):
self.messages.append(str(text))
def show_warning(self, text, parent=None, title=None):
self.warnings.append(str(text))
def show_error(self, text):
self.errors.append(str(text))
def show_critical(self, text):
self.errors.append(str(text))
def update_status(self):
pass
def make_controller(utxos=None):
"""Build a fully-wired BalWindow without constructing the Qt tabs."""
utxos = [make_funding_input()] if utxos is None else utxos
config = FakeConfig()
wallet = FakeWallet(utxos)
window = FakeWindow(wallet)
plugin = BalPlugin(None, config, "bal")
plugin.get_window_title = lambda title: str(title)
plugin.get_decimal_point = window.get_decimal_point
plugin.NO_WILLEXECUTOR.set(True)
plugin.AUTO_REBUILD.set(True)
ctl = BalWindow.__new__(BalWindow)
ctl.bal_plugin = plugin
ctl.window = window
ctl.wallet = wallet
ctl.will = {}
ctl.willitems = {}
ctl.willexecutors = {}
ctl.will_settings = plugin.WILL_SETTINGS.get()
Util.fix_will_settings_tx_fees(ctl.will_settings)
ctl.heirs = Heirs(wallet)
ctl.heirs["alice"] = [ADDRESS, "100000", "1y"]
ctl.heirs["bob"] = [ADDRESS, "100%", "1y"]
ctl.no_willexecutor = True
ctl.disable_plugin = False
ctl.ok = True
ctl.update_all = lambda: None
ctl._schedule_history_refresh = lambda: None
ctl._auto_rebuild_running = False
ctl._auto_rebuild_cooldown_until = 0.0
return ctl
def _no_willexecutors():
"""Force an empty will-executor list (offline tests)."""
return mock.patch.object(
Willexecutors,
"get_willexecutors",
return_value={},
)
def _single(controller):
"""Return (txid, WillItem) for the controller's single will item."""
assert len(controller.willitems) == 1, controller.willitems
return next(iter(controller.willitems.items()))
def _item_spending(controller, *prevout_hexes):
"""Return the will item whose tx spends exactly the given prevouts."""
wanted = sorted(h for h in prevout_hexes)
items = [
item
for item in controller.willitems.values()
if sorted(i.prevout.txid.hex() for i in item.tx.inputs()) == wanted
]
assert len(items) == 1, controller.willitems
return items[0]
# --------------------------------------------------------------------------- #
# Config key
# --------------------------------------------------------------------------- #
def test_auto_rebuild_config_defaults_off():
"""bal_auto_rebuild must default to OFF (False) when not yet stored."""
cfg = FakeConfig()
rebuild = BalConfig(cfg, CONFIG_KEY, False)
assert rebuild.get() is False
def test_auto_rebuild_config_can_be_enabled():
"""Once enabled and persisted, bal_auto_rebuild reads back True."""
cfg = FakeConfig()
rebuild = BalConfig(cfg, CONFIG_KEY, False)
rebuild.set(True)
assert rebuild.get() is True
assert BalConfig(cfg, CONFIG_KEY, False).get() is True
# --------------------------------------------------------------------------- #
# Event wiring (Plugin._wallet_activity)
# --------------------------------------------------------------------------- #
def test_wallet_activity_schedules_only_matching_wallet():
plugin = Plugin.__new__(Plugin)
plugin.AUTO_REBUILD = BalConfig(FakeConfig(), CONFIG_KEY, True)
wallet_a = FakeWallet([make_funding_input()])
wallet_b = FakeWallet([make_funding_input()])
scheduled = []
class _Win:
wallet = wallet_a
ok = True
disable_plugin = False
def schedule_auto_rebuild(self):
scheduled.append(self)
win = _Win()
plugin.bal_windows = {"a": win}
plugin._wallet_activity(wallet_b)
assert scheduled == [], "a different wallet must not schedule a rebuild"
plugin._wallet_activity(wallet_a)
assert scheduled == [win], "the matching wallet must schedule a rebuild"
def test_wallet_activity_skips_when_disabled():
plugin = Plugin.__new__(Plugin)
plugin.AUTO_REBUILD = BalConfig(FakeConfig(), CONFIG_KEY, False)
wallet_obj = FakeWallet([make_funding_input()])
scheduled = []
class _Win:
wallet = wallet_obj
ok = True
disable_plugin = False
def schedule_auto_rebuild(self):
scheduled.append(self)
plugin.bal_windows = {"a": _Win()}
plugin._wallet_activity(wallet_obj)
assert scheduled == [], "AUTO_REBUILD off must not schedule anything"
# --------------------------------------------------------------------------- #
# Scheduling / guards
# --------------------------------------------------------------------------- #
def test_schedule_auto_rebuild_debounces():
ctl = make_controller()
with mock.patch.object(window_mod.QTimer, "singleShot") as single_shot:
ctl.schedule_auto_rebuild()
single_shot.assert_called_once_with(
ctl._AUTO_REBUILD_DEBOUNCE_MS, ctl._run_auto_rebuild
)
def test_auto_rebuild_guards():
with _no_willexecutors():
ctl = make_controller()
ctl.prepare_will()
assert ctl._auto_rebuild_allowed() is True
# Re-entrancy guard.
ctl._auto_rebuild_running = True
assert ctl._auto_rebuild_allowed() is False
ctl._auto_rebuild_running = False
# Cooldown guard.
ctl._auto_rebuild_cooldown_until = time.time() + 100
assert ctl._auto_rebuild_allowed() is False
ctl._auto_rebuild_cooldown_until = 0.0
assert ctl._auto_rebuild_allowed() is True
# Disabled / inactive guards.
ctl.disable_plugin = True
assert ctl._auto_rebuild_allowed() is False
ctl.disable_plugin = False
ctl.ok = False
assert ctl._auto_rebuild_allowed() is False
def test_run_auto_rebuild_spawns_worker_when_allowed():
with _no_willexecutors():
ctl = make_controller()
ctl.prepare_will()
started = []
class FakeThread:
def __init__(self, target, daemon=None):
self.target = target
def start(self):
started.append(self.target)
with mock.patch.object(window_mod.threading, "Thread", FakeThread):
ctl._run_auto_rebuild()
assert len(started) == 1, "the worker thread must be spawned"
# --------------------------------------------------------------------------- #
# maybe_auto_rebuild behaviour
# --------------------------------------------------------------------------- #
def test_auto_rebuild_noop_when_disabled():
with _no_willexecutors():
ctl = make_controller()
ctl.prepare_will()
ctl.bal_plugin.AUTO_REBUILD.set(False)
txid_before, _ = _single(ctl)
result = ctl.maybe_auto_rebuild()
assert result is False
txid_after, _ = _single(ctl)
assert txid_after == txid_before, "disabled flow must not touch the will"
def test_auto_rebuild_noop_without_will():
with _no_willexecutors():
ctl = make_controller()
assert not ctl.willitems
result = ctl.maybe_auto_rebuild()
assert result is False
def test_auto_rebuild_noop_when_will_valid():
with _no_willexecutors():
ctl = make_controller()
ctl.prepare_will()
txid_before, _ = _single(ctl)
with mock.patch.object(ctl, "_auto_invalidate_will") as inv, mock.patch.object(
ctl, "_auto_sign_save_push"
) as sign:
result = ctl.maybe_auto_rebuild()
assert result is False
txid_after, _ = _single(ctl)
assert txid_after == txid_before, "a valid will must not be rebuilt"
inv.assert_not_called()
sign.assert_not_called()
def test_auto_rebuild_rebuilds_and_pushes_on_new_utxo():
with _no_willexecutors():
ctl = make_controller()
# A relative delivery recipe keeps the will coherent after the rebuild
# anticipates the locktime by one day (an absolute recipe would read the
# anticipated tx as a postpone, see check_willexecutors_and_heirs).
ctl.will_settings["locktime"] = "1y"
ctl.prepare_will()
old_txid, old_item = _single(ctl)
old_locktime = int(old_item.tx.locktime)
# An incoming payment adds a second UTXO -> the will no longer covers
# the whole wallet (NotCompleteWillException).
ctl.wallet._utxos.append(make_funding_input("22" * 32))
with mock.patch.object(ctl, "_auto_invalidate_will") as inv, mock.patch.object(
ctl, "push_transactions_to_willexecutors"
) as push, mock.patch.object(ctl, "_save_will_to_history") as history:
result = ctl.maybe_auto_rebuild()
assert result is True, "a stale will must be rebuilt"
assert inv.call_count == 0, "a plain rebuild must not invalidate on-chain"
push.assert_called_once()
history.assert_called_once()
# The rebuilt will now spends BOTH wallet UTXOs (BAL keeps the previous
# single-input transaction alongside it in the will).
new_item = _item_spending(ctl, "11" * 32, "22" * 32)
assert new_item.tx.txid() != old_txid, "the rebuilt will must replace the old tx"
# The new locktime must be at most the old one, so the new tx can be mined
# before the previous will.
assert int(new_item.tx.locktime) <= old_locktime
assert new_item.get_status("COMPLETE"), "passwordless rebuild must sign"
assert new_item.get_status("VALID")
# The rebuilt will is still valid now: no further work.
assert ctl.check_will() is True
def test_auto_rebuild_invalidates_when_threshold_passed():
with _no_willexecutors():
ctl = make_controller()
ctl.prepare_will()
# ADVANCED mode with a check-alive threshold already in the past.
ctl.bal_plugin.USER_TYPE.set("advanced")
ctl.will_settings["threshold"] = int(time.time()) - 3600
with mock.patch.object(ctl, "_auto_invalidate_will") as inv, mock.patch.object(
ctl, "_auto_sign_save_push"
) as sign:
result = ctl.maybe_auto_rebuild()
assert result is True
inv.assert_called_once()
sign.assert_not_called()
def test_auto_rebuild_invalidates_when_locktime_expired():
with _no_willexecutors():
ctl = make_controller()
ctl.prepare_will()
txid, item = _single(ctl)
# Move the frozen delivery date into the past: "too late to
# anticipate" -> the old will must be invalidated on-chain.
item.tx.locktime = int(time.time()) - 2 * 86400
with mock.patch.object(ctl, "_auto_invalidate_will") as inv, mock.patch.object(
ctl, "_auto_sign_save_push"
) as sign:
result = ctl.maybe_auto_rebuild()
assert result is True
inv.assert_called_once()
sign.assert_not_called()
def test_auto_rebuild_invalidates_when_anticipation_crosses_threshold():
with _no_willexecutors():
ctl = make_controller()
now = time.time()
delivery = int(now + 3 * 86400)
ctl.will_settings["locktime"] = delivery
# ADVANCED mode: the check-alive threshold sits 12h before delivery, so
# an anticipated (delivery - 1 day) locktime falls BEFORE it.
ctl.bal_plugin.USER_TYPE.set("advanced")
ctl.will_settings["threshold"] = delivery - 12 * 3600
ctl.prepare_will()
old_txid, _ = _single(ctl)
ctl.wallet._utxos.append(make_funding_input("22" * 32))
# The rebuild itself anticipates the delivery date by one day ONLY when
# the rebuilt transactions keep the same real amounts (Will.check_anticipate,
# same coins + same heirs). Real amounts are re-computed against the
# wallet balance, so a new UTXO normally changes them and the rebuilt
# will keeps the old locktime. Force the anticipating branch here to
# exercise the "anticipated locktime crosses the threshold" handling.
with mock.patch.object(
Will, "check_anticipate", return_value=delivery - 86400
):
with mock.patch.object(
ctl, "_auto_invalidate_will"
) as inv, mock.patch.object(ctl, "_auto_sign_save_push") as sign:
result = ctl.maybe_auto_rebuild()
assert result is True
inv.assert_called_once(), (
"an anticipated locktime below the threshold must invalidate on-chain"
)
sign.assert_not_called(), (
"after an invalidation the rebuilt will must NOT be signed/pushed "
"(the wizard stops and waits for the invalidation to confirm)"
)
new_item = _item_spending(ctl, "11" * 32, "22" * 32)
assert new_item.tx.txid() != old_txid, "the rebuilt will must replace the old tx"
assert int(new_item.tx.locktime) == delivery - 86400, (
"the rebuilt locktime must be anticipated by one day"
)
def _run_all():
tests = [fn for name, fn in sorted(globals().items()) if name.startswith("test_")]
for fn in tests:
print(f"{fn.__name__} ... ", end="", flush=True)
fn()
print("OK")
print(f"\n{len(tests)} tests passed")
if __name__ == "__main__":
app = QApplication.instance() or QApplication([])
_run_all()

4
tests/test_gui_will_flows.py
View File

@@ -269,7 +269,9 @@ def test_prepare_will_builds_and_persists():
assert item.get_status("VALID"), "fresh items default to VALID"
assert txid == item.tx.txid()
assert isinstance(txid, str) and txid.startswith("2"), "raw tx id expected"
assert isinstance(txid, str) and len(txid) == 64 and all(
c in "0123456789abcdef" for c in txid
), "raw tx id expected (64-char hex, not a label/short id)"
assert not item.tx.is_complete(), "unsigned will must not be complete"
assert isinstance(item.tx.locktime, int) and item.tx.locktime > 0