core+gui+cli: animated QR will transfer (balqr/UR1/UR2/BBQR codecs, audio channel, export/import wizard)

This commit is contained in:
2026-09-09 08:43:32 -04:00
parent f45d3be321
commit b3624fef1c
30 changed files with 4037 additions and 624 deletions

View File

@@ -22,7 +22,6 @@ This module is imported lazily (only when a ``bal_*`` command actually runs),
so a missing wallet or a network-less daemon can still start Electrum. so a missing wallet or a network-less daemon can still start Electrum.
""" """
import copy
import json import json
import time import time
@@ -47,7 +46,7 @@ from ..core.checkalive import (
) )
from ..core.heirs import Heirs, is_op_return_address from ..core.heirs import Heirs, is_op_return_address
from ..core.plugin_base import BalConfig, BalPlugin from ..core.plugin_base import BalConfig, BalPlugin
from ..core.util import Util from ..core.util import Util, copy_structure
from ..core.will import Will, WillItem from ..core.will import Will, WillItem
from ..core.willexecutors import Willexecutors, is_onion_url, is_tor_active from ..core.willexecutors import Willexecutors, is_onion_url, is_tor_active
@@ -346,11 +345,11 @@ class BalController:
tx["my_locktime"] = txs[txid].my_locktime tx["my_locktime"] = txs[txid].my_locktime
tx["heirsvalue"] = txs[txid].heirsvalue tx["heirsvalue"] = txs[txid].heirsvalue
tx["description"] = txs[txid].description tx["description"] = txs[txid].description
tx["willexecutor"] = copy.deepcopy(txs[txid].willexecutor) tx["willexecutor"] = copy_structure(txs[txid].willexecutor)
tx["status"] = _("New") tx["status"] = _("New")
tx["baltx_fees"] = txs[txid].tx_fees tx["baltx_fees"] = txs[txid].tx_fees
tx["time"] = creation_time tx["time"] = creation_time
tx["heirs"] = copy.deepcopy(txs[txid].heirs) tx["heirs"] = copy_structure(txs[txid].heirs)
tx["txchildren"] = [] tx["txchildren"] = []
will[txid] = WillItem(tx, _id=txid, wallet=self.wallet) will[txid] = WillItem(tx, _id=txid, wallet=self.wallet)
Will.update_will(self.willitems, will) Will.update_will(self.willitems, will)

1178
bal/core/animated_qr.py Normal file

File diff suppressed because it is too large Load Diff

View File

@@ -59,7 +59,7 @@ from electrum.util import (
write_json_file, write_json_file,
) )
from .util import Util from .util import Util, copy_structure
from .willexecutors import Willexecutors from .willexecutors import Willexecutors
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -321,40 +321,14 @@ def get_change_output(wallet, in_amount, out_amount, fee):
return out return out
def _json_safe(value, _path="heirs", _depth=0): def _json_safe(value, _path="heirs"):
"""Return a JSON-serializable deep copy of *value*. """Backward-compatible alias of :func:`bal.core.util.copy_structure`.
The wallet DB persists the heirs dict via ``json_db.put``, which calls Kept so call sites that imported ``_json_safe`` directly keep working; the
``copy.deepcopy`` on the value. If any nested element is a live runtime actual implementation (a JSON-safe, deepcopy-free clone) lives in
object (e.g. one holding a ``threading.RLock``), deepcopy raises ``bal.core.util`` so every copy path shares one code base.
``TypeError: cannot pickle '_thread.RLock' object`` and the whole
"Build will" task fails.
To make persistence robust we coerce the structure to plain
JSON-compatible types (dict / list / str / int / float / bool / None).
Anything else is converted to ``str(value)`` and logged with its path so
the offending field can be identified, instead of crashing the task.
""" """
# Primitive JSON scalars are kept as-is. return copy_structure(value, _path=_path)
if value is None or isinstance(value, (bool, int, float, str)):
return value
if isinstance(value, dict):
return {
str(k): _json_safe(v, "{}[{!r}]".format(_path, k), _depth + 1)
for k, v in value.items()
}
if isinstance(value, (list, tuple)):
return [
_json_safe(v, "{}[{}]".format(_path, i), _depth + 1)
for i, v in enumerate(value)
]
# Unexpected runtime object: do not let it reach deepcopy. Log where it
# was found so the real source can be fixed, then store a safe string.
_logger.error(
"heirs.save: non-serializable value at {} (type={}); coercing to str. "
"value={!r}".format(_path, type(value).__name__, value)
)
return str(value)
class Heirs(dict, Logger): class Heirs(dict, Logger):

View File

@@ -24,7 +24,7 @@ This module performs **no** GUI work and imports nothing from PyQt / electrum.gu
import json import json
import os import os
import platform import platform
from datetime import date, datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from electrum import constants, json_db from electrum import constants, json_db
from electrum.logging import get_logger from electrum.logging import get_logger
@@ -109,7 +109,9 @@ def get_will(x):
try: try:
# Electrum >= 4.8.0 # Electrum >= 4.8.0
from electrum.stored_dict import register_name as _electrum_register_name # pyright: ignore[reportMissingImports] from electrum.stored_dict import (
register_name as _electrum_register_name, # pyright: ignore[reportMissingImports]
)
def _register_will_dict(name, method, _type=None): def _register_will_dict(name, method, _type=None):
"""Register a plugin dict in the wallet DB (Electrum >= 4.8.0 API).""" """Register a plugin dict in the wallet DB (Electrum >= 4.8.0 API)."""

View File

@@ -21,8 +21,11 @@ import bisect
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from electrum.address_synchronizer import TX_HEIGHT_FUTURE, TX_HEIGHT_LOCAL from electrum.address_synchronizer import TX_HEIGHT_FUTURE, TX_HEIGHT_LOCAL
from electrum.logging import get_logger
from electrum.transaction import PartialTxOutput from electrum.transaction import PartialTxOutput
_logger = get_logger(__name__)
# Bitcoin consensus rule: an nLockTime value strictly below this threshold is # Bitcoin consensus rule: an nLockTime value strictly below this threshold is
# interpreted as a *block height*, otherwise it is interpreted as a *UNIX # interpreted as a *block height*, otherwise it is interpreted as a *UNIX
# timestamp*. # timestamp*.
@@ -35,6 +38,41 @@ from electrum.transaction import PartialTxOutput
LOCKTIME_THRESHOLD = 500000000 LOCKTIME_THRESHOLD = 500000000
def copy_structure(value, _path="copy"):
"""Return a JSON-serializable deep copy of *value*.
This is the ad-hoc, deepcopy-free stand-in used every time the plugin needs
an independent copy of a plain-data structure (heirs dicts, will-executor
dicts, status tables). It recursively clones dict / list / tuple values
while leaving JSON scalars (str / int / float / bool / None) as-is.
If any nested element is a live runtime object (e.g. one holding a
``threading.RLock``), ``copy.deepcopy`` would raise
``TypeError: cannot pickle '_thread.RLock' object``; instead we coerce the
offending value to ``str(value)`` and log it with its path so the source
field can be identified, without crashing the caller.
"""
# Primitive JSON scalars are kept as-is.
if value is None or isinstance(value, (bool, int, float, str)):
return value
if isinstance(value, dict):
return {
str(k): copy_structure(v, "{}[{!r}]".format(_path, k))
for k, v in value.items()
}
if isinstance(value, (list, tuple)):
return [
copy_structure(v, "{}[{}]".format(_path, i)) for i, v in enumerate(value)
]
# Unexpected runtime object: do not let it reach deepcopy. Log where it
# was found so the real source can be fixed, then store a safe string.
_logger.error(
"copy_structure: non-serializable value at {} (type={}); coercing to "
"str. value={!r}".format(_path, type(value).__name__, value)
)
return str(value)
class Util: class Util:
"""Namespace of static helpers (kept as a class to preserve the original """Namespace of static helpers (kept as a class to preserve the original
``Util.method(...)`` call sites used throughout the plugin).""" ``Util.method(...)`` call sites used throughout the plugin)."""

View File

@@ -26,7 +26,6 @@ The status flags themselves (the source of truth) stay here; only the mapping
"status -> colour" now lives in the GUI layer. No behaviour changed. "status -> colour" now lives in the GUI layer. No behaviour changed.
""" """
import copy
from datetime import datetime, timezone from datetime import datetime, timezone
from electrum.i18n import _ from electrum.i18n import _
@@ -45,7 +44,7 @@ from electrum.util import (
) )
from .heirs import WillExecutorFeeTooHighException from .heirs import WillExecutorFeeTooHighException
from .util import Util from .util import Util, copy_structure
from .willexecutors import Willexecutors from .willexecutors import Willexecutors
MIN_LOCKTIME = 1 MIN_LOCKTIME = 1
@@ -143,7 +142,7 @@ class Will:
willitems = {} willitems = {}
for wid in will: for wid in will:
Will.add_info_from_will(will, wid, wallet) Will.add_info_from_will(will, wid, wallet)
willitems[wid] = WillItem(will[wid]) willitems[wid] = WillItem(will[wid], wallet=wallet)
will = willitems will = willitems
errors = {} errors = {}
for wid in will: for wid in will:
@@ -165,7 +164,7 @@ class Will:
outputs = will[wid].tx.outputs() outputs = will[wid].tx.outputs()
ow = will[wid] ow = will[wid]
ow.normalize_locktime(others_input) ow.normalize_locktime(others_input)
will[wid] = WillItem(ow.to_dict()) will[wid] = ow.copy()
for i in range(0, len(outputs)): for i in range(0, len(outputs)):
Will.change_input( Will.change_input(
@@ -465,7 +464,7 @@ class Will:
continue continue
utxo_str = utxo.prevout.to_str() utxo_str = utxo.prevout.to_str()
if utxo_str in prevout_to_spend: if utxo_str in prevout_to_spend:
balance += inputs[utxo_str][0][2].value_sats() balance += utxo.value_sats()
utxo_to_spend.append(utxo) utxo_to_spend.append(utxo)
_logger.debug("utxo to spend: {}".format(utxo_to_spend)) _logger.debug("utxo to spend: {}".format(utxo_to_spend))
if len(utxo_to_spend) > 0: if len(utxo_to_spend) > 0:
@@ -1327,14 +1326,20 @@ class WillItem(Logger):
return self.STATUS[status][1] return self.STATUS[status][1]
def __init__(self, w, _id=None, wallet=None): def __init__(self, w, _id=None, wallet=None):
if isinstance( if isinstance(w, WillItem):
w, # Copy a WillItem WITHOUT deepcopy. Serialize it to its plain-dict
WillItem, # form and deserialize from there: the tx is re-parsed into a fresh
): # object, STATUS is rebuilt from the clones below and heirs /
self.__dict__ = w.__dict__.copy() # will-executors are cloned recursively, so the copy shares no
self.STATUS = copy.deepcopy(w.STATUS) # mutable state with the source. See also copy().
self.heirs = copy.deepcopy(w.heirs) if w.heirs is not None else None data = w.to_dict()
else: data["heirs"] = copy_structure(w.heirs) if w.heirs is not None else None
data["willexecutor"] = (
copy_structure(w.we) if w.we is not None else None
)
if not _id:
_id = w._id
w = data
self.tx = Will.get_tx_from_any(w["tx"]) self.tx = Will.get_tx_from_any(w["tx"])
self.heirs = w.get("heirs", None) self.heirs = w.get("heirs", None)
self.we = w.get("willexecutor", None) self.we = w.get("willexecutor", None)
@@ -1347,7 +1352,7 @@ class WillItem(Logger):
self.sigs_have = int(w.get("sigs_have", 0)) self.sigs_have = int(w.get("sigs_have", 0))
self.father = w.get("Father", None) self.father = w.get("Father", None)
self.children = w.get("Children", None) self.children = w.get("Children", None)
self.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT) self.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
for s in self.STATUS: for s in self.STATUS:
self.STATUS[s][1] = w.get(s, WillItem.STATUS_DEFAULT[s][1]) self.STATUS[s][1] = w.get(s, WillItem.STATUS_DEFAULT[s][1])
# Backward-compatibility migration (A2): the "PENDING" status was # Backward-compatibility migration (A2): the "PENDING" status was
@@ -1370,6 +1375,27 @@ class WillItem(Logger):
if wallet: if wallet:
self.tx.add_info_from_wallet(wallet) self.tx.add_info_from_wallet(wallet)
def copy(self, wallet=None):
"""Return an independent copy of this WillItem (no deepcopy).
The copy is produced by serializing this item and deserializing it:
the transaction is re-parsed, the STATUS table is rebuilt and
heirs / will-executors are cloned recursively, so the result shares no
mutable state with ``self``. Pass a ``wallet`` when the copy's tx
needs its address/value information restored
(``tx.add_info_from_wallet``).
"""
return WillItem(self, _id=self._id, wallet=wallet)
@staticmethod
def copy_status_table(status_table):
"""Clone a STATUS table (``{flag: [label, bool]}``) without deepcopy.
Both the outer dict and every inner ``[label, bool]`` list are new
objects, so mutating the returned table never affects the source.
"""
return {k: [label, value] for k, (label, value) in status_table.items()}
def to_dict(self): def to_dict(self):
out = { out = {
"_id": self._id, "_id": self._id,
@@ -1383,6 +1409,8 @@ class WillItem(Logger):
"baltx_fees": self.tx_fees, "baltx_fees": self.tx_fees,
"sigs_required": self.sigs_required, "sigs_required": self.sigs_required,
"sigs_have": self.sigs_have, "sigs_have": self.sigs_have,
"Father": self.father,
"Children": self.children,
} }
for key in self.STATUS: for key in self.STATUS:
try: try:

View File

@@ -16,11 +16,10 @@ the Qt button and the OS/subprocess glue.
import os import os
import subprocess import subprocess
from electrum.gui.qt.util import getSaveFileName
from PyQt6.QtGui import QAction from PyQt6.QtGui import QAction
from PyQt6.QtWidgets import QInputDialog, QMenu, QToolButton from PyQt6.QtWidgets import QInputDialog, QMenu, QToolButton
from electrum.gui.qt.util import getSaveFileName
from ...core.reminders import write_temp_ics from ...core.reminders import write_temp_ics
from .common import _, _logger from .common import _, _logger

View File

@@ -15,7 +15,6 @@ hosts a few GUI helpers that do not deserve a module of their own:
(:class:`CheckAliveError` now lives in ``bal.core.checkalive``.) (:class:`CheckAliveError` now lives in ``bal.core.checkalive``.)
""" """
import copy
import enum import enum
import os import os
import subprocess import subprocess
@@ -82,6 +81,7 @@ from PyQt6.QtWidgets import (
QAbstractItemView, QAbstractItemView,
QAbstractSpinBox, QAbstractSpinBox,
QApplication, QApplication,
QButtonGroup,
QCheckBox, QCheckBox,
QComboBox, QComboBox,
QDateTimeEdit, QDateTimeEdit,
@@ -94,6 +94,7 @@ from PyQt6.QtWidgets import (
QMenu, QMenu,
QMenuBar, QMenuBar,
QPushButton, QPushButton,
QRadioButton,
QScrollArea, QScrollArea,
QSizePolicy, QSizePolicy,
QSpinBox, QSpinBox,
@@ -121,7 +122,7 @@ from ...core.heirs import (
# --- Core (GUI-free) logic layer --- # --- Core (GUI-free) logic layer ---
from ...core.plugin_base import BalPlugin, BalTimestamp from ...core.plugin_base import BalPlugin, BalTimestamp
from ...core.util import Util from ...core.util import Util, copy_structure
from ...core.will import ( from ...core.will import (
AmountException, AmountException,
HeirChangeException, HeirChangeException,

File diff suppressed because it is too large Load Diff

View File

@@ -20,15 +20,13 @@ from PyQt6.QtWidgets import QLineEdit as _QLineEdit
from PyQt6.QtWidgets import QMessageBox, QStyledItemDelegate from PyQt6.QtWidgets import QMessageBox, QStyledItemDelegate
from .common import ( from .common import (
_, OP_RETURN_PREFIX,
_logger,
BalTimestamp, BalTimestamp,
Buttons, Buttons,
CancelButton, CancelButton,
HelpButton, HelpButton,
MessageBoxMixin, MessageBoxMixin,
MyTreeView, MyTreeView,
OP_RETURN_PREFIX,
OkButton, OkButton,
QAbstractItemView, QAbstractItemView,
QApplication, QApplication,
@@ -46,15 +44,17 @@ from .common import (
QSpinBox, QSpinBox,
QStandardItem, QStandardItem,
QStandardItemModel, QStandardItemModel,
Qt,
QToolButton, QToolButton,
QVBoxLayout, QVBoxLayout,
QWidget, QWidget,
Qt,
TaskThread, TaskThread,
Util, Util,
Will, Will,
WillItem,
Willexecutors, Willexecutors,
WillItem,
_,
_logger,
char_width_in_lineedit, char_width_in_lineedit,
datetime, datetime,
enum, enum,
@@ -63,8 +63,8 @@ from .common import (
import_meta_gui, import_meta_gui,
is_op_return_address, is_op_return_address,
partial, partial,
read_QIcon_from_bytes,
read_json_file, read_json_file,
read_QIcon_from_bytes,
server_status_text, server_status_text,
server_status_tooltip, server_status_tooltip,
signature_suffix, signature_suffix,
@@ -663,13 +663,11 @@ class PreviewList(MyTreeView, MessageBoxMixin):
menu.addAction(_("Prepare"), self.build_transactions) menu.addAction(_("Prepare"), self.build_transactions)
menu.addAction(_("Display"), self.bal_window.preview_modal_dialog) menu.addAction(_("Display"), self.bal_window.preview_modal_dialog)
menu.addAction(_("Sign"), self.ask_password_and_sign_transactions) menu.addAction(_("Sign"), self.ask_password_and_sign_transactions)
export_menu = menu.addMenu(_("Export")) # Export/Import open a single window that offers all transports
export_menu.addAction(_("All"), self.export_will) # (file / QR / audio). The Choose Filter / transport settings live
export_menu.addAction(_("Valid"), self.export_will_valid) # inside that window.
export_menu.addAction(_("Valid NC"), self.export_will_valid_incomplete) menu.addAction(_("Export"), self.export_will)
export_menu.addAction(_("QR Codes"), self.export_will_via_qr) menu.addAction(_("Import"), self.import_will)
menu.addAction(_("Import"), self.import_will_into_details)
menu.addAction(_("Import via QR"), self.import_will_via_qr)
menu.addAction(_("Merge"), self.merge_will) menu.addAction(_("Merge"), self.merge_will)
menu.addAction(_("Broadcast"), self.broadcast) menu.addAction(_("Broadcast"), self.broadcast)
menu.addAction(_("Check"), self.check) menu.addAction(_("Check"), self.check)
@@ -735,48 +733,15 @@ class PreviewList(MyTreeView, MessageBoxMixin):
if will: if will:
self.update_will(will) self.update_will(will)
def export_json_file(self, path):
write_json_file(path, self.will)
def export_will(self): def export_will(self):
self.bal_window.export_will() self.bal_window.export_will_dialog()
self.update()
def export_will_valid(self): def import_will(self):
"""Export only the will items that are valid.""" self.bal_window.import_will_dialog()
subset = {
wid: wi
for wid, wi in self.will.items()
if wi.get_status("VALID")
}
if not subset:
self.show_message(_("No valid will item to export"))
return
self.bal_window.export_will(will=subset)
self.update()
def export_will_valid_incomplete(self):
"""Export only the will items that are valid but not yet fully signed (V-NC)."""
subset = {
wid: wi
for wid, wi in self.will.items()
if wi.get_status("VALID") and not wi.get_status("COMPLETE")
}
if not subset:
self.show_message(_("No valid, incomplete will item to export"))
return
self.bal_window.export_will(will=subset)
self.update()
def import_will_into_details(self): def import_will_into_details(self):
self.bal_window.import_will_into_details() self.bal_window.import_will_into_details()
def export_will_via_qr(self):
self.bal_window.export_will_via_qr()
def import_will_via_qr(self):
self.bal_window.import_will_via_qr()
def merge_will(self): def merge_will(self):
self.bal_window.merge_will_ui() self.bal_window.merge_will_ui()

View File

@@ -20,10 +20,7 @@ from electrum.util import EventListener, event_listener
from PyQt6.QtWidgets import QLayout from PyQt6.QtWidgets import QLayout
from ...core.qrtransfer import CHUNK_PRESETS, preset_index_for_chunk_size from ...core.qrtransfer import CHUNK_PRESETS, preset_index_for_chunk_size
from .common import ( from .common import (
_,
_logger,
BalPlugin, BalPlugin,
Buttons, Buttons,
EnterButton, EnterButton,
@@ -40,6 +37,8 @@ from .common import (
QWidget, QWidget,
UserCancelled, UserCancelled,
Willexecutors, Willexecutors,
_,
_logger,
add_widget, add_widget,
partial, partial,
read_QIcon_from_bytes, read_QIcon_from_bytes,

View File

@@ -29,17 +29,15 @@ from ...core.input_rules import (
from ...core.reminders import build_ics_reminders, write_temp_ics from ...core.reminders import build_ics_reminders, write_temp_ics
from .calendar import BalCalendar, BalCalendarButton from .calendar import BalCalendar, BalCalendarButton
from .common import ( from .common import (
_,
_logger,
Any,
BTCAmountEdit,
BalTimestamp,
ColorScheme,
DECIMAL_POINT, DECIMAL_POINT,
Decimal,
HelpButton,
NLOCKTIME_BLOCKHEIGHT_MAX, NLOCKTIME_BLOCKHEIGHT_MAX,
NLOCKTIME_MAX, NLOCKTIME_MAX,
Any,
BalTimestamp,
BTCAmountEdit,
ColorScheme,
Decimal,
HelpButton,
Optional, Optional,
QAbstractSpinBox, QAbstractSpinBox,
QCheckBox, QCheckBox,
@@ -57,13 +55,15 @@ from .common import (
QSpinBox, QSpinBox,
QStyle, QStyle,
QStyleOptionFrame, QStyleOptionFrame,
Qt,
QTextEdit, QTextEdit,
QVBoxLayout, QVBoxLayout,
QWidget, QWidget,
Qt,
Union, Union,
Util, Util,
Will, Will,
_,
_logger,
char_width_in_lineedit, char_width_in_lineedit,
datetime, datetime,
getSaveFileName, getSaveFileName,

View File

@@ -25,8 +25,7 @@ from ...core.checkalive import (
resolve_date_to_check, resolve_date_to_check,
) )
from .common import ( from .common import (
_, OP_RETURN_PREFIX,
_logger,
AmountException, AmountException,
BalPlugin, BalPlugin,
Buttons, Buttons,
@@ -40,10 +39,10 @@ from .common import (
Mapping, Mapping,
Network, Network,
NoHeirsException, NoHeirsException,
NoWillExecutorNotPresent,
NotCompleteWillException, NotCompleteWillException,
OP_RETURN_PREFIX, NoWillExecutorNotPresent,
OkButton, OkButton,
Optional,
PaymentIdentifier, PaymentIdentifier,
QGridLayout, QGridLayout,
QLabel, QLabel,
@@ -57,15 +56,17 @@ from .common import (
TxFeesChangedException, TxFeesChangedException,
Util, Util,
Will, Will,
WillexecutorChangeException,
WillExecutorFeeTooHighException, WillExecutorFeeTooHighException,
WillExecutorNotPresent, WillExecutorNotPresent,
Willexecutors,
WillExpiredException, WillExpiredException,
WillItem, WillItem,
WillPostponedException, WillPostponedException,
WillexecutorChangeException, _,
Willexecutors, _logger,
char_width_in_lineedit, char_width_in_lineedit,
copy, copy_structure,
export_meta_gui, export_meta_gui,
import_meta_gui, import_meta_gui,
is_onion_url, is_onion_url,
@@ -73,8 +74,8 @@ from .common import (
is_tor_active, is_tor_active,
log_error, log_error,
partial, partial,
read_QIcon_from_bytes,
read_json_file, read_json_file,
read_QIcon_from_bytes,
show_on_top, show_on_top,
shown_cv, shown_cv,
time, time,
@@ -88,8 +89,10 @@ from .dialogs import (
BalWizardDialog, BalWizardDialog,
WillDetailDialog, WillDetailDialog,
WillExecutorDialog, WillExecutorDialog,
WillQrExportDialog, WillExportDialog,
WillQrImportDialog, WillImportDialog,
_complete_import,
decode_will_payload,
) )
from .lists import HeirListWidget, PreviewList from .lists import HeirListWidget, PreviewList
from .widgets import LockTimeWidget, PercAmountEdit from .widgets import LockTimeWidget, PercAmountEdit
@@ -517,11 +520,11 @@ class BalWindow:
tx["my_locktime"] = txs[txid].my_locktime tx["my_locktime"] = txs[txid].my_locktime
tx["heirsvalue"] = txs[txid].heirsvalue tx["heirsvalue"] = txs[txid].heirsvalue
tx["description"] = txs[txid].description tx["description"] = txs[txid].description
tx["willexecutor"] = copy.deepcopy(txs[txid].willexecutor) tx["willexecutor"] = copy_structure(txs[txid].willexecutor)
tx["status"] = _("New") tx["status"] = _("New")
tx["baltx_fees"] = txs[txid].tx_fees tx["baltx_fees"] = txs[txid].tx_fees
tx["time"] = creation_time tx["time"] = creation_time
tx["heirs"] = copy.deepcopy(txs[txid].heirs) tx["heirs"] = copy_structure(txs[txid].heirs)
tx["txchildren"] = [] tx["txchildren"] = []
will[txid] = WillItem(tx, _id=txid, wallet=self.wallet) will[txid] = WillItem(tx, _id=txid, wallet=self.wallet)
self.update_will(will) self.update_will(will)
@@ -1638,6 +1641,19 @@ class BalWindow:
else: else:
write_json_file(path, {wid: wi.to_dict() for wid, wi in will.items()}) write_json_file(path, {wid: wi.to_dict() for wid, wi in will.items()})
def export_tx_file(self, path, will=None):
"""Export only the serialized transactions of the given will items.
Writes a plain text file with every transaction (or PSBT) serialized
on a single line, separated by a comma (``tx1,tx2,tx3``). The raw hex
and PSBT base64 alphabets never contain a comma, so the separator is
unambiguous. When ``will`` is omitted the live will items are used.
"""
willitems = will if will is not None else self.willitems
serialized = ",".join(str(wi.tx) for wid, wi in willitems.items())
with open(path, "w", encoding="utf-8") as f:
f.write(serialized)
def export_will(self, will=None): def export_will(self, will=None):
try: try:
export_meta_gui( export_meta_gui(
@@ -1647,18 +1663,23 @@ class BalWindow:
self.show_error(str(e)) self.show_error(str(e))
raise e raise e
def export_will_via_qr(self, will=None): def export_will_dialog(self, will=None, initial_mode: Optional[str] = None):
"""Export the will (default: the live one) as QR codes on screen. """Open the unified export window (File / QR / Audio).
The selected transactions are serialized with the wire format of The window lets the user pick an All / Valid / Valid NC filter in
:mod:`bal.core.qrtransfer` and shown, one frame at a time, in a the top row and choose one of the three transports, each with its
:class:`WillQrExportDialog`. When Electrum's ``audio_modem`` plugin contextual settings (file format for File, QR-code size and autoplay
is available (:meth:`get_audio_modem_plugin`) the dialog also offers for QR, KB/sec for Audio). ``will`` defaults to the live will items;
an "Audio" send button. ``initial_mode`` opens the window directly on the given transport.
""" """
try: try:
willitems = will if will is not None else self.willitems willitems = will if will is not None else self.willitems
d = WillQrExportDialog(self, will=willitems, bal_plugin=self.bal_plugin) d = WillExportDialog(
self,
will=willitems,
bal_plugin=self.bal_plugin,
initial_mode=initial_mode or "file",
)
show_on_top(d) show_on_top(d)
except Exception as e: except Exception as e:
self.show_error(str(e)) self.show_error(str(e))
@@ -1693,6 +1714,22 @@ class BalWindow:
return return
plugin._send(parent=self.window, blob=payload) plugin._send(parent=self.window, blob=payload)
def set_audio_modem_bitrate(self, kbps):
"""Set the ``audio_modem`` plugin transfer speed to ``kbps`` KB/sec.
Both the send and the receive paths read ``modem_config``, so the
sender and the receiver must be configured with the same speed. Raises
when the plugin (or its ``amodem`` dependency) is unavailable.
"""
plugin = self.get_audio_modem_plugin()
if plugin is None:
raise Exception(_("Audio MODEM plugin is not available."))
try:
import amodem.config
except Exception as e:
raise Exception(str(e)) from e
plugin.modem_config = amodem.config.bitrates[int(kbps)]
def merge_will(self, imported): def merge_will(self, imported):
"""Merge imported will items into the live will. """Merge imported will items into the live will.
@@ -1816,16 +1853,34 @@ class BalWindow:
def on_file(path): def on_file(path):
try: try:
willitems = self._load_will_file(path) with open(path, "r", encoding="utf-8") as f:
text = f.read()
except Exception as e: except Exception as e:
self.show_error(_("Invalid will file: {}").format(e)) self.show_error(_("Invalid will file: {}").format(e))
return return
# Attach wallet/input info so the imported txs can be signed and kind, data = decode_will_payload(text)
# broadcast (mirrors what merge_will_from_file does). try:
if kind == "will":
willitems = self._load_will_payload(data)
# Attach wallet/input info so the imported txs can be
# signed and broadcast (mirrors merge_will_from_file).
Will.normalize_will(willitems, self.wallet) Will.normalize_will(willitems, self.wallet)
for wi in willitems.values(): for wi in willitems.values():
wi.set_status("IMPORTED", True) wi.set_status("IMPORTED", True)
imported.update(willitems) imported.update(willitems)
else:
# Serialized transactions: route through the shared import
# tail (validity pass + review/sign wizard).
_complete_import(
self,
self.bal_plugin,
text,
show_error=self.show_error,
show_warning=self.show_warning,
close=lambda: None,
)
except Exception as e:
self.show_error(_("Invalid will file: {}").format(e))
def on_success(): def on_success():
if not imported: if not imported:
@@ -1835,16 +1890,16 @@ class BalWindow:
import_meta_gui(self.window, _("will"), on_file, on_success) import_meta_gui(self.window, _("will"), on_file, on_success)
def import_will_via_qr(self): def import_will_dialog(self):
"""Import a will through QR codes (or audio) and review/sign it. """Open the unified import window (File / QR / Audio).
Opens a :class:`WillQrImportDialog`. The captured transactions are The window offers three transports: File opens the read-only
parsed into fresh :class:`WillItem` objects (never touching the :class:`WillDetailDialog` preview; QR and Audio capture the
live will), run through the same local validity pass the merge flow transfer and send it through the per-transaction review wizard
uses, and are then presented in the per-transaction review wizard (:class:`WillTxReviewSignDialog`). Every flow works on fresh
(:class:`WillTxReviewSignDialog`). :class:`WillItem` objects and never touches the live will.
""" """
d = WillQrImportDialog(self, bal_plugin=self.bal_plugin) d = WillImportDialog(self, bal_plugin=self.bal_plugin)
show_on_top(d) show_on_top(d)
def _load_will_file(self, path): def _load_will_file(self, path):
@@ -1855,6 +1910,15 @@ class BalWindow:
willitems[k] = WillItem(data[k], _id=k) willitems[k] = WillItem(data[k], _id=k)
return willitems return willitems
def _load_will_payload(self, data):
"""Build WillItems from decoded whole-will JSON data."""
willitems = {}
for k, v in data.items():
d = dict(v)
d["tx"] = tx_from_any(d["tx"])
willitems[k] = WillItem(d, _id=k)
return willitems
def check_transactions_task(self, will): def check_transactions_task(self, will):
start = time.time() start = time.time()
# Servers are now contacted in parallel (see # Servers are now contacted in parallel (see

View File

@@ -6,6 +6,9 @@ target-version = "py312"
select =["E", "W", "F", "I", "N", "B"] select =["E", "W", "F", "I", "N", "B"]
ignore = ["E501"] ignore = ["E501"]
[tool.ruff.lint.pep8-naming]
classmethod-decorators = ["classmethod", "classproperty"] # electrum.util.classproperty uses cls
[tool.ruff.lint.per-file-ignores] [tool.ruff.lint.per-file-ignores]
"bal/gui/qt/common.py" = ["F401"] # re-exports consumed via explicit imports "bal/gui/qt/common.py" = ["F401"] # re-exports consumed via explicit imports
"bal/gui/qt/dialogs.py" = ["N802"] # Qt overrides: closeEvent/hideEvent/getText "bal/gui/qt/dialogs.py" = ["N802"] # Qt overrides: closeEvent/hideEvent/getText

View File

@@ -21,12 +21,12 @@ Run:
QT_QPA_PLATFORM=offscreen PYTHONPATH=electrum-src python3 tests/sim_update_flows.py QT_QPA_PLATFORM=offscreen PYTHONPATH=electrum-src python3 tests/sim_update_flows.py
""" """
import copy
import os import os
import sys import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
from bal.core.util import copy_structure
from bal.core.will import ( from bal.core.will import (
HeirNotFoundException, HeirNotFoundException,
NoHeirsException, NoHeirsException,
@@ -58,7 +58,7 @@ def _make_will_item(heirs, tx_locktime, status_complete=False):
is forced to ``tx_locktime`` (the value frozen in the signed Bitcoin tx).""" is forced to ``tx_locktime`` (the value frozen in the signed Bitcoin tx)."""
d = { d = {
"tx": _VALID_TX_HEX, "tx": _VALID_TX_HEX,
"heirs": copy.deepcopy(heirs), "heirs": copy_structure(heirs),
"willexecutor": None, "willexecutor": None,
"status": "", "status": "",
"description": "", "description": "",
@@ -67,7 +67,7 @@ def _make_will_item(heirs, tx_locktime, status_complete=False):
"baltx_fees": TX_FEES, "baltx_fees": TX_FEES,
} }
item = WillItem(d, _id="willid_1") item = WillItem(d, _id="willid_1")
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT) item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
# Force the locktime frozen "inside" the signed tx. # Force the locktime frozen "inside" the signed tx.
item.tx.locktime = tx_locktime item.tx.locktime = tx_locktime
if status_complete: if status_complete:
@@ -118,7 +118,7 @@ def main():
# Scenario 0: nothing changed -> should be coherent. # Scenario 0: nothing changed -> should be coherent.
heirs = {"alice": ["addr_alice", 5000, same_lt]} heirs = {"alice": ["addr_alice", 5000, same_lt]}
_run("0. nothing changed", _run("0. nothing changed",
will_heirs=heirs, current_heirs=copy.deepcopy(heirs), will_heirs=heirs, current_heirs=copy_structure(heirs),
tx_locktime=base_lt, check_date=0) tx_locktime=base_lt, check_date=0)
# Scenario 1: delivery date moved forward (postpone), will NOT yet signed. # Scenario 1: delivery date moved forward (postpone), will NOT yet signed.

View File

@@ -27,7 +27,6 @@ Run:
tests/test_anticipate_manual_locktime.py -q tests/test_anticipate_manual_locktime.py -q
""" """
import copy
import os import os
import sys import sys
@@ -35,6 +34,7 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
import pytest # noqa: E402 # pyright: ignore[reportMissingImports] import pytest # noqa: E402 # pyright: ignore[reportMissingImports]
from bal.core.util import copy_structure # noqa: E402
from bal.core.will import ( # noqa: E402 from bal.core.will import ( # noqa: E402
NotCompleteWillException, NotCompleteWillException,
Will, Will,
@@ -70,7 +70,7 @@ def _make_will_item(heirs, tx_locktime, status_complete=False):
""" """
d = { d = {
"tx": _VALID_TX_HEX, "tx": _VALID_TX_HEX,
"heirs": copy.deepcopy(heirs), "heirs": copy_structure(heirs),
"willexecutor": None, "willexecutor": None,
"status": "", "status": "",
"description": "", "description": "",
@@ -79,7 +79,7 @@ def _make_will_item(heirs, tx_locktime, status_complete=False):
"baltx_fees": TX_FEES, "baltx_fees": TX_FEES,
} }
item = WillItem(d, _id="willid_1") item = WillItem(d, _id="willid_1")
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT) item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
item.tx.locktime = tx_locktime item.tx.locktime = tx_locktime
if status_complete: if status_complete:
item.set_status("COMPLETE", True) item.set_status("COMPLETE", True)

View File

@@ -0,0 +1,478 @@
"""
Tests for ``bal.core.animated_qr`` (BC-UR v1, BC-UR v2, BBQR interop).
Validates the self-contained codecs against the published spec vectors
(BCR-2020-004/005 BC32, BCR-2020-012 bytewords) and against byte-exact
output captured from the reference C++ bc-ur encoder (fountain/xoshiro/
alias-sampler parity), plus round trips, out-of-order assembly, missing-part
fountain solving and malformed-input rejection for all four formats.
Run:
source electrum/env/bin/activate
python3 tests/test_core_animated_qr.py
"""
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
import random
from bal.core import animated_qr as aq
def _payload(plen: int) -> bytes:
"""Deterministic payload matching the C++ reference driver (``(i*7)&0xff``)."""
return bytes((i * 7) & 0xFF for i in range(plen))
# --------------------------------------------------------------------------- #
# BC32 (BCR-2020-004 / bcr-2020-005 rev1 reference implementation vectors)
# --------------------------------------------------------------------------- #
def test_bc32_official_vectors():
cases = [
(b"Hello, world", "fpjkcmr09ss8wmmjd3jq6ax7w9"),
(b"Hello world", "fpjkcmr0ypmk7unvvsh4ra4j"),
(
bytes.fromhex("d934063e82001eec0585ee41ab5d8e4b703a4be1f73aec21e143912c56"),
"my6qv05zqq0wcpv9aeq6khvwfdcr5jlp7uawcg0pgwgjc4shjm6xu",
),
]
for payload, encoded in cases:
assert aq.bc32_encode(payload) == encoded
assert aq.bc32_decode(encoded) == payload
def test_bc32_checksum_rejected():
good = aq.bc32_encode(b"Hello, world")
corrupted = good[:-1] + ("a" if good[-1] != "a" else "b")
try:
aq.bc32_decode(corrupted)
except aq.AnimatedQrError:
pass
else:
raise AssertionError("expected AnimatedQrError for corrupted BC32")
def test_bc32_bad_char_rejected():
try:
aq.bc32_decode("1" * 26)
except aq.AnimatedQrError:
pass
else:
raise AssertionError("expected AnimatedQrError for '1' (not in alphabet)")
# --------------------------------------------------------------------------- #
# Bytewords (BCR-2020-012)
# --------------------------------------------------------------------------- #
def test_bytewords_minimal_roundtrip():
samples = [bytes(range(256)), _payload(59), b"\x00"] + [
os.urandom(64) for _ in range(4)
]
for data in samples:
words = aq.bytewords_minimal_encode(data)
assert len(words) == (len(data) + 4) * 2 # 2 chars per byte incl. CRC
assert aq.bytewords_minimal_decode(words) == data
def test_bytewords_rejects_corrupted_crc():
data = _payload(40)
words = aq.bytewords_minimal_encode(data)
flip = "a" if words[-1] != "a" else "b"
try:
aq.bytewords_minimal_decode(words[:-1] + flip)
except aq.AnimatedQrError:
pass
else:
raise AssertionError("expected AnimatedQrError for corrupted CRC")
def test_bytewords_rejects_odd_length():
try:
aq.bytewords_minimal_decode("abc")
except aq.AnimatedQrError:
pass
else:
raise AssertionError("expected AnimatedQrError for odd-length bytewords")
# --------------------------------------------------------------------------- #
# BC-UR v2: byte-exact parity with the reference C++ encoder
# --------------------------------------------------------------------------- #
# Reference frames from the bc-ur C++ fountain encoder
# (payload x=(i*7)&0xFF, cbor wrapped, single-part and multipart).
REF_V2_SINGLE_12 = "ur:bytes/gsaeatbabzcecndrehetfhfggtoeemhpmo"
REF_V2_MULTI_59 = [
"ur:bytes/2-2/lpaoaocsfscyrpdpjzbyhdctsbtdtavtvdwyykztaxbkbycsctdsdpeefrfwgagdhghyihjzjkknlylomymtaeeccasket",
"ur:bytes/3-2/lpaxaocsfscyrpdpjzbyhdcthdfraeatbabzcecndrehetfhfggtghhpidinjoktkblplkmunyoypdperpryssimryrldt",
"ur:bytes/4-2/lpaaaocsfscyrpdpjzbyhdctsbtdtavtvdwyykztaxbkbycsctdsdpeefrfwgagdhghyihjzjkknlylomymtaefeimteue",
"ur:bytes/5-2/lpahaocsfscyrpdpjzbyhdctmuwltavdwlzowlurdtfrdtdihkjekkjlhkdnesdidtuywlzmwluydtdiesdnssgdaontls",
"ur:bytes/6-2/lpamaocsfscyrpdpjzbyhdctmuwltavdwlzowlurdtfrdtdihkjekkjlhkdnesdidtuywlzmwluydtdiesdnssisescmwt",
"ur:bytes/7-2/lpataocsfscyrpdpjzbyhdcthdfraeatbabzcecndrehetfhfggtghhpidinjoktkblplkmunyoypdperprysslsspplgm",
"ur:bytes/8-2/lpayaocsfscyrpdpjzbyhdctsbtdtavtvdwyykztaxbkbycsctdsdpeefrfwgagdhghyihjzjkknlylomymtaeonlrzebg",
"ur:bytes/9-2/lpasaocsfscyrpdpjzbyhdctsbtdtavtvdwyykztaxbkbycsctdsdpeefrfwgagdhghyihjzjkknlylomymtaeaaryknzt",
"ur:bytes/10-2/lpbkaocsfscyrpdpjzbyhdctmuwltavdwlzowlurdtfrdtdihkjekkjlhkdnesdidtuywlzmwluydtdiesdnsslotsfrfn",
"ur:bytes/11-2/lpbdaocsfscyrpdpjzbyhdctmuwltavdwlzowlurdtfrdtdihkjekkjlhkdnesdidtuywlzmwluydtdiesdnssdtwyrstd",
"ur:bytes/12-2/lpbnaocsfscyrpdpjzbyhdctsbtdtavtvdwyykztaxbkbycsctdsdpeefrfwgagdhghyihjzjkknlylomymtaegswnvdin",
"ur:bytes/13-2/lpbtaocsfscyrpdpjzbyhdctmuwltavdwlzowlurdtfrdtdihkjekkjlhkdnesdidtuywlzmwluydtdiesdnsshknlptee",
]
# Reference message for the 59-byte payload: byte-string head (0x58,0x3b) + data.
REF_V2_MULTI_59_MSG = bytes([0x58, 0x3B]) + _payload(59)
def test_v2_single_part_matches_reference():
frames = aq.ur2_frames(_payload(12), len(REF_V2_SINGLE_12))
assert frames == [REF_V2_SINGLE_12]
def test_v2_reference_frames_decode_and_reencode_exactly():
message = REF_V2_MULTI_59_MSG
fragment_len = -(-len(message) // 2)
for frame in REF_V2_MULTI_59:
seq, seq_len, message_len, checksum, data = aq.ur2_parse_part(frame)
assert seq_len == 2
assert message_len == len(message)
assert checksum == aq.crc32_int(message)
assert len(data) == fragment_len
# re-encoding the parsed values reproduces the reference line exactly
assert aq._ur2_part_string(seq, seq_len, message_len, checksum, data) == frame
# our choose_fragments + partition + xor reproduces the reference data
indexes = aq.choose_fragments(seq, seq_len, checksum)
assert seq_num_indexes_valid(seq, seq_len, indexes)
mixed = aq._mix_fragments(aq._partition_message(message, fragment_len), indexes, fragment_len)
assert mixed == data
def seq_num_indexes_valid(seq, seq_len, indexes):
# pure part for seq <= seq_len contains exactly fragment seq-1
if seq <= seq_len:
return indexes == {seq - 1}
return set(indexes) <= set(range(seq_len)) and bool(indexes)
def test_v2_multipart_encoder_matches_reference_from_seq2():
# Our frames start at seq 1 (spec-aligned); parts seq 2.. must equal the
# reference (which starts at seq 2 due to first_seq_num=1).
mine = aq.ur2_frames(_payload(59), 120)
assert mine[0].split("/", 1)[1].startswith("1-2") or "1-2" in mine[0].split("/")[1]
assert mine[1:4] == REF_V2_MULTI_59[:3]
def test_v2_reference_seq7_mix_parity():
# Higher-degree mixed parts (seq_len=7) also match: message uses the
# reference head 0x58|0x00 for the 256-byte driver payload.
message = bytes([0x58, 0x00]) + _payload(256)
seq_len = 7
fragment_len = -(-len(message) // seq_len)
frames = [
"ur:bytes/9-7/lpasatcfadaocyfysnjlsrhddaykztaxbkbycsctdsdpeefrfwgagdhghyihjzjkknlylomymtntoxpyprrhrtsttotluovlwdwnsrfejzhd",
"ur:bytes/10-7/lpbkatcfadaocyfysnjlsrhddazeahbnbwcycldedlenfsfygrgmhkhniojtkpkelslememkneolpmqzrksasotitsuevwwpwfzswzpmdrvo",
"ur:bytes/11-7/lpbdatcfadaocyfysnjlsrhddawkwtbbbefnaefnbebbjojybebnaebndybbbewkwtceaecedyeebebbjobnaebnbeeedybbbeztwproyapd",
]
for frame in frames:
seq, sl, mlen, checksum, data = aq.ur2_parse_part(frame)
assert sl == seq_len and mlen == len(message)
assert checksum == aq.crc32_int(message)
mixed = aq._mix_fragments(
aq._partition_message(message, fragment_len),
aq.choose_fragments(seq, seq_len, checksum),
fragment_len,
)
assert mixed == data
# --------------------------------------------------------------------------- #
# BC-UR v2: sessions / fountain decoding
# --------------------------------------------------------------------------- #
def test_v2_roundtrip_in_order():
payload = ("BAL transfer " * 9).encode()
frames = aq.ur2_frames(payload, 120)
seq_len = int(frames[0].split("/")[1].split("-")[1])
assert len(frames) == 2 * seq_len # pure wave + redundant mixed wave
session = aq.AnimatedQrSession()
for frame in frames:
session.add_part(frame)
assert session.done
assert session.received == session.total
text, _ = session.resolve()
assert text == payload.decode()
def test_v2_out_of_order_and_duplicate():
payload = ("BAL transfer " * 9).encode()
frames = aq.ur2_frames(payload, 120)
order = list(range(len(frames)))
random.Random(11).shuffle(order)
session = aq.AnimatedQrSession()
for i in order:
status = session.add_part(frames[i])
assert status in ("ok", "dup")
session.add_part(frames[0]) # duplicate of an already-received part
assert session.done
assert session.resolve()[0] == payload.decode()
def test_v2_solves_without_a_pure_fragment():
payload = ("BAL transfer " * 9).encode()
frames = aq.ur2_frames(payload, 120)
session = aq.AnimatedQrSession()
for frame in frames[1:]: # drop the first pure fragment
session.add_part(frame)
assert session.done
assert session.resolve()[0] == payload.decode()
def test_v2_single_part_import():
session = aq.AnimatedQrSession()
session.add_part(REF_V2_SINGLE_12)
assert session.done and session.total == 1
assert session.resolve()[0] == _payload(12).decode("latin-1")
def test_v2_conflicting_transfer_rejected():
payload_a = b"AAAAAAAAAAAAAAAA"
payload_b = b"BBBBBBBBBBBBBBBB"
fa = aq.ur2_frames(payload_a, 500)[0]
fb = aq.ur2_frames(payload_b, 500)[0]
session = aq.AnimatedQrSession()
session.add_part(fa)
try:
session.add_part(fb)
except aq.TransferConflictError:
pass
else:
raise AssertionError("expected TransferConflictError for a different transfer")
def test_v2_corrupt_crc_rejected():
frame = list(REF_V2_MULTI_59[0])
idx = len(frame) - 1
frame[idx] = "a" if frame[idx] != "a" else "b"
try:
aq.ur2_parse_part("".join(frame))
except aq.AnimatedQrError:
pass
else:
raise AssertionError("expected AnimatedQrError for a corrupt v2 part")
def test_v2_session_cap_rejected():
part = aq._ur2_part_string(1, 30000, 100, 1234, b"\x00" * 100)
session = aq._Ur2Session()
try:
session.add(part)
except aq.SessionLimitError:
pass
else:
raise AssertionError("expected SessionLimitError for oversized seq_len")
# --------------------------------------------------------------------------- #
# BC-UR v1
# --------------------------------------------------------------------------- #
def test_v1_multipart_roundtrip():
payload = ("v1 transfer payload " * 6).encode()
frames = aq.ur1_frames(payload, 120)
assert len(frames) > 1
session = aq.AnimatedQrSession()
for frame in reversed(frames):
session.add_part(frame)
assert session.done
assert session.resolve()[0] == payload.decode()
def test_v1_single_part_roundtrip():
payload = b"hello, bal"
frames = aq.ur1_frames(payload, 400)
assert len(frames) == 1
session = aq.AnimatedQrSession()
session.add_part(frames[0])
assert session.done and session.total == 1
assert session.resolve()[0] == payload.decode()
def test_v1_headerless_single_part_import():
# bcr-2020-005 rev1 allows omitting the sequence header + digest entirely.
payload = b"hello, bal"
message = aq.cbor_byte_string(payload)
single = "ur:bytes/" + aq.bc32_encode(message)
assert aq.detect_format(single) == "ur1"
session = aq.AnimatedQrSession()
session.add_part(single)
assert session.done
assert session.resolve()[0] == payload.decode()
def test_v1_digest_mismatch_rejected():
frame = aq.ur1_frames(b"hello, bal", 400)[0]
tampered = frame[:-4] + "abcd"
session = aq.AnimatedQrSession()
session.add_part(tampered)
try:
session.resolve()
except aq.ChecksumError:
pass
else:
raise AssertionError("expected ChecksumError for a tampered v1 digest")
def test_v1_part_numbers_validated():
for bad in (
"ur:bytes/0of1/{}full".format("x" * 51),
"ur:bytes/2of1/{}full".format("x" * 51),
"ur:bytes/1of0/{}full".format("x" * 51),
"ur:bytes/1aof1/{}full".format("x" * 51),
):
try:
aq.ur1_parse_part(bad)
except aq.AnimatedQrError:
pass
else:
raise AssertionError("expected AnimatedQrError for: {}".format(bad))
# --------------------------------------------------------------------------- #
# BBQR
# --------------------------------------------------------------------------- #
def test_bbqr_all_encodings_roundtrip():
payload = ("BBQR payload " * 8).encode()
for encoding in ("Z", "2", "H"):
frames = aq.bbqr_frames(payload, 90, encoding=encoding)
assert len(frames) >= 1
order = list(range(len(frames)))
random.Random(3).shuffle(order)
session = aq.AnimatedQrSession()
for i in order:
session.add_part(frames[i])
assert session.done
assert session.resolve()[0] == payload.decode()
def test_bbqr_compression_default_and_fallback():
payload = ("repetitive data " * 40).encode() # compresses well
frames_z = aq.bbqr_frames(payload, 90, encoding="Z")
# Highly compressible: Z yields one frame and a 'Z' flag.
assert all(f[2] == "Z" for f in frames_z)
assert len(frames_z) == 1
raw = os.urandom(600) # incompressible
frames_2 = aq.bbqr_frames(raw, 90, encoding="Z")
assert all(f[2] == "2" for f in frames_2) # Z loses, '2' is used
def test_bbqr_hex_uppercase():
payload = b"\xde\xad\xbe\xef"
frame = aq.bbqr_frames(payload, 50, encoding="H")[0]
assert "DEADBEEF" in frame
encoding, _type, total, index, frag = aq.bbqr_parse_part(frame)
assert (encoding, total, index) == ("H", 1, 0)
def test_bbqr_runt_last_part():
payload = os.urandom(33)
frames = aq.bbqr_frames(payload, 60, encoding="2")
parts = [aq.bbqr_parse_part(f)[4] for f in frames]
joined = aq._bbqr_decode(parts, "2")
assert joined == payload
assert len(parts[-1]) < len(parts[0]) # last part is a runt
def test_bbqr_zlib_bomb_rejected():
compressed = aq._bbqr_encode(b"\x00" * 1000000, "Z")[1]
try:
aq._bbqr_decode(["0" * len(compressed)], "2") # not zlib data
except aq.AnimatedQrError:
pass
# direct inflate bomb guard:
inflated = aq._bbqr_encode(b"\x00" * 1000000, "Z")
assert inflated[0] == "Z" # 1MB zeros compresses
bomb = aq._bbqr_encode(b"\x00" * (aq._MAX_MESSAGE_BYTES + 100), "Z")[1]
parts = [bomb[i : i + 90] for i in range(0, len(bomb), 90)]
try:
aq._bbqr_decode(parts, "Z")
except aq.AnimatedQrError:
pass
else:
raise AssertionError("expected AnimatedQrError for an oversized decompression")
def test_bbqr_part_number_limits():
try:
aq.bbqr_frames(os.urandom(30000), 40, encoding="2")
except aq.AnimatedQrError:
pass
else:
raise AssertionError("expected AnimatedQrError for too many BBQR parts")
# --------------------------------------------------------------------------- #
# Detection / parse_for_detection
# --------------------------------------------------------------------------- #
def test_detect_format_recognises_all_formats():
assert aq.detect_format("BALQR1|1|1||payload") == "balqr"
assert aq.detect_format(aq.ur1_frames(b"x", 400)[0]) == "ur1"
assert aq.detect_format(aq.ur2_frames(b"x", 400)[0]) == "ur2"
assert aq.detect_format(aq.bbqr_frames(b"x", 50)[0]) == "bbqr"
assert aq.detect_format(REF_V2_SINGLE_12) == "ur2"
assert aq.detect_format("ur:bytes/" + aq.bc32_encode(aq.cbor_byte_string(b"x"))) == "ur1"
def test_detect_format_rejects_garbage():
for text in ("", "hello world", "BALQ|1|1||a", "ur:", "ur:txn/xyz"):
assert aq.detect_format(text) is None, text
# Lenient prefix probe: a string that merely *starts* with "balqr" is
# reported as balqr (the strict parse then rejects it downstream).
assert aq.detect_format("BALQRX|1|1||a") == "balqr"
def test_parse_for_detection_keys():
bal = aq.parse_for_detection("BALQR1|3|2||payload")
assert bal == ("balqr", "balqr:3", 3, 2)
v2 = aq.parse_for_detection(aq.ur2_frames(b"x"*50, 400)[0])
assert v2[0] == "ur2" and v2[2] == 1 and v2[3] == 1
v1 = aq.parse_for_detection(aq.ur1_frames(b"x"*50, 120)[0])
assert v1[0] == "ur1" and v1[2] > 1 and 1 <= v1[3] <= v1[2]
bb = aq.parse_for_detection(aq.bbqr_frames(b"x"*50, 40)[0])
assert bb[0] == "bbqr" and bb[2] >= 1 and 0 <= bb[3] < bb[2]
def test_format_names_exist():
for fmt in ("balqr", "ur1", "ur2", "bbqr"):
assert aq.format_name(fmt)
assert aq.format_name("nope") == "nope"
# --------------------------------------------------------------------------- #
if __name__ == "__main__":
import traceback
failures = 0
for _name, fn in sorted(globals().items()):
if _name.startswith("test_") and callable(fn):
try:
fn()
print("ok: {}".format(_name))
except Exception:
failures += 1
print("FAIL: {}".format(_name))
traceback.print_exc()
if failures:
print("{} test(s) failed".format(failures))
sys.exit(1)
print("all tests passed")

View File

@@ -14,7 +14,7 @@ import time
sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
from datetime import date, datetime, timedelta from datetime import date, datetime, timedelta, timezone
from bal.core.plugin_base import BalConfig, BalPlugin, BalTimestamp from bal.core.plugin_base import BalConfig, BalPlugin, BalTimestamp
@@ -74,7 +74,7 @@ def test_bt_to_date_absolute():
def test_bt_to_date_relative(): def test_bt_to_date_relative():
now = datetime.now() now = datetime.now(timezone.utc)
# relative days from now # relative days from now
bt = BalTimestamp("7d") bt = BalTimestamp("7d")
@@ -86,8 +86,8 @@ def test_bt_to_date_relative():
d_rev = bt.to_date(reverse=True) d_rev = bt.to_date(reverse=True)
assert d_rev < now assert d_rev < now
# from explicit datetime # from explicit datetime (UTC, so the naive-timestamp roundtrip below is stable)
base = datetime(2025, 6, 1, 12, 0, 0) base = datetime(2025, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
d = bt.to_date(from_date=base) d = bt.to_date(from_date=base)
expected = (base + timedelta(days=7)).replace(hour=0, minute=0, second=0, microsecond=0) expected = (base + timedelta(days=7)).replace(hour=0, minute=0, second=0, microsecond=0)
assert d == expected assert d == expected
@@ -101,7 +101,7 @@ def test_bt_to_date_relative():
def test_bt_to_date_years(): def test_bt_to_date_years():
bt = BalTimestamp("1y") bt = BalTimestamp("1y")
d = bt.to_date() d = bt.to_date()
assert d > datetime.now() assert d > datetime.now(timezone.utc)
def test_bt_to_date_overflow(): def test_bt_to_date_overflow():

View File

@@ -293,7 +293,6 @@ def test_presets_fit_qrcode_ec_m():
"""Every preset budget must render inside a QR at EC level M.""" """Every preset budget must render inside a QR at EC level M."""
try: try:
import qrcode import qrcode
from qrcode.constants import ERROR_CORRECT_M from qrcode.constants import ERROR_CORRECT_M
except ImportError: except ImportError:
print("qrcode not installed - skipping capacity check") print("qrcode not installed - skipping capacity check")

View File

@@ -8,12 +8,12 @@ Run:
python3 tests/test_core_will.py python3 tests/test_core_will.py
""" """
import copy
import os import os
import sys import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
from bal.core.util import copy_structure
from bal.core.will import Will, WillItem from bal.core.will import Will, WillItem
# A valid serialized Bitcoin transaction hex (1 input + 1 P2PKH output, version 2) # A valid serialized Bitcoin transaction hex (1 input + 1 P2PKH output, version 2)
@@ -48,7 +48,7 @@ def _make_willitem_blank():
"""Create a fresh WillItem from scratch.""" """Create a fresh WillItem from scratch."""
item = WillItem(_make_minimal_willitem_dict()) item = WillItem(_make_minimal_willitem_dict())
# Reset STATUS to clean defaults # Reset STATUS to clean defaults
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT) item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
return item return item
@@ -151,8 +151,8 @@ def test_will_only_valid_list():
def _make_will_with_heirs(heirs, tx_locktime): def _make_will_with_heirs(heirs, tx_locktime):
"""Build a single-item will whose stored heirs == ``heirs`` and whose """Build a single-item will whose stored heirs == ``heirs`` and whose
frozen tx.locktime == ``tx_locktime`` (what the will-executors hold).""" frozen tx.locktime == ``tx_locktime`` (what the will-executors hold)."""
item = WillItem(_make_minimal_willitem_dict(heirs=copy.deepcopy(heirs))) item = WillItem(_make_minimal_willitem_dict(heirs=copy_structure(heirs)))
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT) item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
item.tx.locktime = tx_locktime item.tx.locktime = tx_locktime
return {"willid_1": item} return {"willid_1": item}
@@ -163,7 +163,7 @@ def test_check_heirs_unchanged_is_coherent():
heirs = {"alice": ["addr_alice", 5000, str(lt)]} heirs = {"alice": ["addr_alice", 5000, str(lt)]}
will = _make_will_with_heirs(heirs, lt) will = _make_will_with_heirs(heirs, lt)
result = Will.check_willexecutors_and_heirs( result = Will.check_willexecutors_and_heirs(
will, copy.deepcopy(heirs), {}, False, 0, 100 will, copy_structure(heirs), {}, False, 0, 100
) )
assert result is True assert result is True

View File

@@ -17,7 +17,6 @@ Run:
python3 -m pytest tests/test_core_will_invalidate.py -q python3 -m pytest tests/test_core_will_invalidate.py -q
""" """
import copy
import os import os
import sys import sys
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
@@ -75,7 +74,7 @@ def _make_willitem(value_sats=1000000, valid=True, extra_heirs=None):
"change": "", "change": "",
"baltx_fees": 100, "baltx_fees": 100,
}) })
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT) item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
# Set the input value so the balance calculation works. # Set the input value so the balance calculation works.
# Use the name-mangled attribute because tx_from_any creates a # Use the name-mangled attribute because tx_from_any creates a
# Transaction whose inputs are TxInput objects; TxInput.value_sats() # Transaction whose inputs are TxInput objects; TxInput.value_sats()
@@ -270,7 +269,7 @@ class TestInvalidateWill:
""" """
item = _make_willitem(value_sats=100) item = _make_willitem(value_sats=100)
will = {"willtxid1": item} will = {"willtxid1": item}
wallet = _mock_wallet([_make_utxo()]) wallet = _mock_wallet([_make_utxo(value_sats=100)])
result, mock_from_io, _ = _run_invalidate(will, wallet, fees_per_byte=100) result, mock_from_io, _ = _run_invalidate(will, wallet, fees_per_byte=100)

View File

@@ -21,7 +21,6 @@ Run:
python3 -m pytest tests/test_group_e_karen7_invalidate.py -q python3 -m pytest tests/test_group_e_karen7_invalidate.py -q
""" """
import copy
import json import json
import os import os
import sys import sys
@@ -46,6 +45,7 @@ from electrum.transaction import (
from electrum.util import bfh from electrum.util import bfh
from bal.core.heirs import Heirs from bal.core.heirs import Heirs
from bal.core.util import copy_structure
from bal.core.will import Will, WillItem from bal.core.will import Will, WillItem
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #
@@ -219,7 +219,7 @@ def _txs_to_will(txs, heirs_data):
for txid, tx in txs.items(): for txid, tx in txs.items():
item_dict = { item_dict = {
"tx": tx, "tx": tx,
"heirs": copy.deepcopy(heirs_data), "heirs": copy_structure(heirs_data),
"willexecutor": None, "willexecutor": None,
"status": "", "status": "",
"description": "", "description": "",

View File

@@ -22,7 +22,6 @@ Run:
python3 -m pytest tests/test_group_e_mock_karen7.py -q python3 -m pytest tests/test_group_e_mock_karen7.py -q
""" """
import copy
import json import json
import os import os
import sys import sys
@@ -43,6 +42,7 @@ from bal.core.reminders import (
ical_escape, ical_escape,
write_temp_ics, write_temp_ics,
) )
from bal.core.util import copy_structure
from bal.core.will import HeirNotFoundException, Will, WillItem from bal.core.will import HeirNotFoundException, Will, WillItem
from bal.core.willexecutors import Willexecutors from bal.core.willexecutors import Willexecutors
@@ -136,7 +136,7 @@ def _make_willitem(**overrides):
} }
d.update(overrides) d.update(overrides)
item = WillItem(d) item = WillItem(d)
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT) item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
return item return item
@@ -343,7 +343,7 @@ def test_e2_heir_change_triggers_rebuild():
item = WillItem( item = WillItem(
{ {
"tx": _VALID_TX_HEX, "tx": _VALID_TX_HEX,
"heirs": copy.deepcopy(will_heirs), "heirs": copy_structure(will_heirs),
"willexecutor": None, "willexecutor": None,
"status": "", "status": "",
"description": "", "description": "",
@@ -352,7 +352,7 @@ def test_e2_heir_change_triggers_rebuild():
"baltx_fees": 100, "baltx_fees": 100,
} }
) )
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT) item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
item.tx.locktime = lt item.tx.locktime = lt
will = {"willid_1": item} will = {"willid_1": item}

View File

@@ -0,0 +1,456 @@
"""
Tests for the filter-based unified export/import dialogs and the BalWindow
transport helpers (``bal.gui.qt.dialogs``, ``bal.gui.qt.window``).
Covers the shared export filters (All / Valid / Valid NC), the unified
``WillExportDialog`` file page (whole item vs tx-only content, empty-filter
abort), the audio export/import pages (KB/sec wiring, missing plugin guard,
receive flow) and the comma-separated tx-only file writer. The audio pages
run against a stub plugin so no sound hardware is exercised.
Run:
QT_QPA_PLATFORM=offscreen python3 tests/test_gui_export_dialogs.py
"""
import base64
import json
import sys
import zlib
from unittest.mock import patch
sys.path.insert(0, __file__.rsplit("/", 2)[0])
from PyQt6.QtWidgets import QApplication, QMainWindow
import bal.gui.qt.dialogs as dialogs
from bal.core.qrtransfer import CHUNK_PRESETS
_app = QApplication.instance() or QApplication(sys.argv)
# A valid 1x1 transparent PNG, good enough for BalDialog's window icon.
_PNG_BYTES = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
)
# A valid serialized Bitcoin transaction hex (1 input + 1 P2PKH output,
# version 2); used for real-WillItem serialization tests.
_VALID_TX_HEX = (
"01000000012a5c9a94fcde98f5581cd00162c60a13936ceb75389ea65b"
"f38633b424eb4031000000006c493046022100a82bbc57a0136751e543"
"3f41cf000b3f1a99c6744775e76ec764fb78c54ee100022100f9e80b7d"
"e89de861dc6fb0c1429d5da72c2b6b2ee2406bc9bfb1beedd729d98501"
"2102e61d176da16edd1d258a200ad9759ef63adf8e14cd97f53227bae3"
"5cdb84d2f6ffffffff0140420f00000000001976a914230ac37834073a"
"42146f11ef8414ae929feaafc388ac00000000"
)
class _Cfg:
def __init__(self, value):
self._value = value
def get(self):
return self._value
class FakePlugin:
QR_CHUNK_SIZE = _Cfg(CHUNK_PRESETS[0][1])
def read_file(self, path):
return _PNG_BYTES
class FakeWindow(QMainWindow):
config = {}
def format_amount(self, amount):
return "{:.8f}".format(amount)
def format_amount_and_units(self, amount):
return "{:.8f} sat".format(amount)
class FakeAudioPlugin:
"""Duck-typed ``audio_modem`` plugin for the audio pages."""
def __init__(self):
self.modem_config = None
def is_available(self):
return True
class _FakeModemConfig:
def __init__(self, kbps):
self.modem_bps = kbps * 1000
class FakeBalWindow:
"""Duck-typed stand-in for BalWindow (dialog layer only)."""
def __init__(self, audio_plugin=None):
self.window = FakeWindow()
self.bal_plugin = FakePlugin()
self.willitems = {}
self.audio_plugin = audio_plugin
self.bitrate_set = None
self.audio_payloads = []
def get_audio_modem_plugin(self):
return self.audio_plugin
def set_audio_modem_bitrate(self, kbps):
self.bitrate_set = kbps
if self.audio_plugin is not None:
self.audio_plugin.modem_config = _FakeModemConfig(kbps)
def _audio_send_payload(self, payload):
self.audio_payloads.append(payload)
def export_json_file(self, path, will=None):
items = will if will is not None else self.willitems
with open(path, "w", encoding="utf-8") as f:
json.dump({wid: wi.to_dict() for wid, wi in items.items()}, f)
def export_tx_file(self, path, will=None):
items = will if will is not None else self.willitems
with open(path, "w", encoding="utf-8") as f:
f.write(",".join(str(wi.tx) for _, wi in items.items()))
class StubTx:
def __init__(self, payload):
self.payload = payload
def txid(self):
return "{:064x}".format(hash(self.payload) & 0xFFFFFFFFFFFFFFFF)
def __str__(self):
return self.payload
class StubWillItem:
def __init__(self, payload, statuses=None):
self.tx = StubTx(payload)
self.statuses = statuses or {}
def get_status(self, name):
return self.statuses.get(name, False)
def to_dict(self):
return {"tx": str(self.tx)}
def _make_willitems(n=3, payload_len=60, statuses=None):
return {
"item{}".format(i): StubWillItem(
"T{}".format(i) * payload_len, statuses=statuses
)
for i in range(n)
}
# ------------------------------------------------------------------ #
# Shared export filters
# ------------------------------------------------------------------ #
def test_export_filter_options():
opts = dialogs.export_filter_options()
assert [label for label, _fn in opts] == ["All", "Valid", "Valid NC"]
complete = StubWillItem("C*", statuses={"VALID": True, "COMPLETE": True})
valid = StubWillItem("V*", statuses={"VALID": True})
plain = StubWillItem("P*")
assert opts[0][1](complete) and opts[0][1](plain)
assert opts[1][1](complete) and opts[1][1](valid) and not opts[1][1](plain)
assert not opts[2][1](complete)
assert opts[2][1](valid) and not opts[2][1](plain)
def test_filter_willitems_by_index():
items = {
"a": StubWillItem("A*", statuses={"VALID": True, "COMPLETE": True}),
"b": StubWillItem("B*", statuses={"VALID": True}),
"c": StubWillItem("C*"),
}
opts = dialogs.export_filter_options()
assert set(dialogs.filter_willitems(items, opts, 0)) == {"a", "b", "c"}
assert set(dialogs.filter_willitems(items, opts, 1)) == {"a", "b"}
assert dialogs.filter_willitems(items, opts, 2) == {"b": items["b"]}
# ------------------------------------------------------------------ #
# WillExportDialog file page
# ------------------------------------------------------------------ #
def test_file_export_selects_by_filter():
bw = FakeBalWindow()
items = _make_willitems(3)
items["item0"].statuses = {"VALID": True, "COMPLETE": True}
items["item1"].statuses = {"VALID": True}
d = dialogs.WillExportDialog(bw, will=items, bal_plugin=bw.bal_plugin)
assert set(d._selected_items()) == set(items)
d._on_filter_change(1)
assert set(d._selected_items()) == {"item0", "item1"}
d._on_filter_change(2)
assert list(d._selected_items()) == ["item1"]
d.close()
def test_file_export_run_tx_only(tmpdir):
bw = FakeBalWindow()
items = _make_willitems(3)
d = dialogs.WillExportDialog(bw, will=items, bal_plugin=bw.bal_plugin)
d.content_check.setChecked(False)
captured = {}
def fake_gui(window, title, exporter):
captured["title"] = title
captured["exporter"] = exporter
with patch.object(dialogs, "export_meta_gui", side_effect=fake_gui):
d._export_file()
assert captured["title"] == "will_tx"
out = tmpdir.join("will_tx.txt").strpath
captured["exporter"](out)
expected = ",".join(str(wi.tx) for _, wi in items.items())
assert open(out, encoding="utf-8").read() == expected
d.close()
def test_file_export_run_willitem(tmpdir):
bw = FakeBalWindow()
items = _make_willitems(2)
d = dialogs.WillExportDialog(bw, will=items, bal_plugin=bw.bal_plugin)
assert d.content_check.isChecked()
captured = {}
def fake_gui(window, title, exporter):
captured["title"] = title
captured["exporter"] = exporter
with patch.object(dialogs, "export_meta_gui", side_effect=fake_gui):
d._export_file()
assert captured["title"] == "will"
out = tmpdir.join("will.json").strpath
captured["exporter"](out)
data = json.load(open(out, encoding="utf-8"))
assert set(data) == set(items)
assert data["item0"]["tx"] == str(items["item0"].tx)
d.close()
def test_file_export_empty_under_filter_aborts():
bw = FakeBalWindow()
items = {
"a": StubWillItem("A*", statuses={"VALID": True, "COMPLETE": True})
}
d = dialogs.WillExportDialog(bw, will=items, bal_plugin=bw.bal_plugin)
messages = []
d.show_message = lambda msg: messages.append(msg) # type: ignore[assignment]
d._filter_index = 2 # force an empty "Valid NC" selection
with patch.object(dialogs, "export_meta_gui") as gui:
d._export_file()
assert not gui.called
assert messages
d.close()
# ------------------------------------------------------------------ #
# BalWindow transport helpers
# ------------------------------------------------------------------ #
def test_bal_window_export_tx_file(tmpdir):
import bal.gui.qt.window as window
items = _make_willitems(3)
bw = object.__new__(window.BalWindow)
bw.willitems = items
out = tmpdir.join("will_tx.txt").strpath
bw.export_tx_file(out)
expected = ",".join(str(wi.tx) for _, wi in items.items())
assert open(out, encoding="utf-8").read() == expected
def test_bal_window_set_audio_modem_bitrate():
try:
import amodem.config
except ImportError:
return
import bal.gui.qt.window as window
class P:
def __init__(self):
self.modem_config = None
probe = P()
bw = object.__new__(window.BalWindow)
bw.get_audio_modem_plugin = lambda: probe
bw.set_audio_modem_bitrate(1)
assert probe.modem_config is amodem.config.bitrates[1]
# ------------------------------------------------------------------ #
# WillExportDialog audio page
# ------------------------------------------------------------------ #
def test_audio_export_page_send():
bw = FakeBalWindow(audio_plugin=FakeAudioPlugin())
items = _make_willitems(3, payload_len=30)
bw.willitems = items
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin)
assert d.audio_send_btn.text() == dialogs._("Send")
assert d.audio_send_btn.isEnabled()
assert d.kbps_combo.count() > 0
kbps = int(d.kbps_combo.currentText())
d._send_audio()
assert bw.bitrate_set == kbps
# Whole-will default: the audio payload is a single JSON document.
assert len(bw.audio_payloads) == 1
data = json.loads(bw.audio_payloads[0])
assert set(data) == set(items)
d.close()
def test_audio_export_page_send_tx_only():
bw = FakeBalWindow(audio_plugin=FakeAudioPlugin())
items = _make_willitems(3, payload_len=30)
bw.willitems = items
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin)
d.content_check.setChecked(False)
kbps = int(d.kbps_combo.currentText())
d._send_audio()
assert bw.bitrate_set == kbps
expected = "\n".join(dialogs.serialize_tx_list(items))
assert bw.audio_payloads == [expected]
d.close()
def test_audio_export_page_plugin_missing():
# The window stays usable: only the audio option is disabled.
bw = FakeBalWindow(audio_plugin=None)
bw.willitems = _make_willitems(2)
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin)
assert not d.audio_send_btn.isEnabled()
assert "not available" in d.audio_warn_label.text()
assert len(d.qr_page.frames) >= 1 # QR still usable
assert d.file_export_btn.isEnabled()
d.close()
# ------------------------------------------------------------------ #
# WillImportDialog audio page
# ------------------------------------------------------------------ #
def test_audio_import_page_build():
bw = FakeBalWindow(audio_plugin=FakeAudioPlugin())
d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
assert d.kbps_combo.count() > 0
assert d.receive_btn.text() == dialogs._("Receive by audio…")
assert d.receive_btn.isEnabled()
d.close()
def test_audio_import_page_plugin_missing():
bw = FakeBalWindow(audio_plugin=None)
d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
assert not d.receive_btn.isEnabled()
assert "not available" in d.audio_warn_label.text()
assert d.qr_page is not None # QR import still usable
d.close()
def test_audio_import_receive_wiring():
bw = FakeBalWindow(audio_plugin=FakeAudioPlugin())
d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
captured = {}
class FakeWaitingDialog:
def __init__(self, parent, msg, task, on_success=None, on_error=None):
captured["msg"] = msg
captured["task"] = task
captured["success"] = on_success
captured["error"] = on_error
imported = []
def fake_complete(bal_window, bal_plugin, payload, **kwargs):
imported.append(payload)
blob = zlib.compress(b"A" * 40 + b"\n" + b"B" * 40)
with patch.object(dialogs, "WaitingDialog", FakeWaitingDialog), patch.object(
dialogs, "_complete_import", side_effect=fake_complete
):
d._audio_receive()
kbps = int(d.kbps_combo.currentText())
assert bw.bitrate_set == kbps
assert captured["task"] is not None
captured["success"](blob)
# Payload is the raw decompressed text; autodetect handles the splitting.
assert imported == ["A" * 40 + "\n" + "B" * 40]
d.close()
# ------------------------------------------------------------------ #
# Whole-will JSON payload serializes a real Transaction (MyEncoder)
# ------------------------------------------------------------------ #
def test_qr_whole_will_json_serializes_transaction():
"""Regression: _whole_will_json must not raise
"Object of type Transaction is not JSON serializable".
Real WillItems keep a ``Transaction`` object in ``tx``; the whole-will
QR payload (default content scope) must serialize it via MyEncoder the
same way write_json_file does.
"""
from bal.core.will import WillItem
item = WillItem({
"tx": _VALID_TX_HEX,
"heirs": {},
"willexecutor": None,
"status": "",
"description": "",
"time": 0,
"change": "",
"baltx_fees": 100,
})
bw = FakeBalWindow()
d = dialogs.WillExportDialog(
bw, will={"imp0": item}, bal_plugin=bw.bal_plugin, initial_mode="qr"
)
j = d._whole_will_json()
data = json.loads(j)
assert data["imp0"]["tx"] == _VALID_TX_HEX
d.close()
# ------------------------------------------------------------------ #
# Main
# ------------------------------------------------------------------ #
if __name__ == "__main__":
import inspect
import os
import tempfile
class _Path:
def __init__(self, d, name):
self.strpath = os.path.join(d, name)
class _Tmp:
def __init__(self):
self._d = tempfile.mkdtemp()
def join(self, name):
return _Path(self._d, name)
tmp = _Tmp()
for name in sorted(dir()):
if name.startswith("test_"):
fn = globals()[name]
fn(tmp) if inspect.signature(fn).parameters else fn()
print(" [OK] {}".format(name))
print("[OK] All export dialog GUI tests passed")

View File

@@ -1,19 +1,21 @@
""" """
Tests for the QR / audio will-transfer dialogs (``bal.gui.qt.dialogs``). Tests for the QR / audio will-transfer dialogs (``bal.gui.qt.dialogs``).
Covers WillQrExportDialog (build, frame navigation, chunk-size change) and Covers the unified ``WillExportDialog`` (transport radios, stacked pages,
WillQrImportDialog (frame capture, slot grid, complete-review enabling, total QR build/navigation, chunk-size / autoplay, filter revert) and the unified
mismatch reset, frame assembly -> decode). The wizard and the camera/audio ``WillImportDialog`` (QR frame capture, slot grid, complete-review enabling,
paths need a live wallet/hardware and are exercised only through the shared total mismatch reset, frame assembly -> decode). The wizard and the
frame-assembly path here. camera/audio paths need a live wallet/hardware and are exercised only
through the shared frame-assembly path here.
Run: Run:
QT_QPA_PLATFORM=offscreen python3 tests/test_gui_qr_transfer.py QT_QPA_PLATFORM=offscreen python3 tests/test_gui_qr_transfer.py
""" """
import base64 import base64
import json
import sys import sys
from unittest.mock import patch from unittest.mock import MagicMock, patch
sys.path.insert(0, __file__.rsplit("/", 2)[0]) sys.path.insert(0, __file__.rsplit("/", 2)[0])
@@ -21,7 +23,7 @@ from electrum.transaction import Transaction
from PyQt6.QtWidgets import QApplication, QMainWindow from PyQt6.QtWidgets import QApplication, QMainWindow
import bal.gui.qt.dialogs as dialogs import bal.gui.qt.dialogs as dialogs
from bal.core.qrtransfer import encode_transfer, split_frames from bal.core.qrtransfer import CHUNK_PRESETS, encode_transfer, split_frames
from bal.core.will import WillItem from bal.core.will import WillItem
_app = QApplication.instance() or QApplication(sys.argv) _app = QApplication.instance() or QApplication(sys.argv)
@@ -44,7 +46,18 @@ _VALID_TX_HEX = (
) )
class _Cfg:
def __init__(self, value):
self._value = value
def get(self):
return self._value
class FakePlugin: class FakePlugin:
# Smallest QR preset: long transfers produce several frames.
QR_CHUNK_SIZE = _Cfg(CHUNK_PRESETS[0][1])
def read_file(self, path): def read_file(self, path):
return _PNG_BYTES return _PNG_BYTES
@@ -90,6 +103,9 @@ class StubWillItem:
def get_status(self, name): def get_status(self, name):
return self.statuses.get(name, False) return self.statuses.get(name, False)
def to_dict(self):
return {"tx": str(self.tx), "status": self.statuses}
def _make_willitems(n=3, payload_len=120): def _make_willitems(n=3, payload_len=120):
return { return {
@@ -99,30 +115,61 @@ def _make_willitems(n=3, payload_len=120):
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #
# WillQrExportDialog # WillExportDialog (QR transport via d.qr_page)
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #
def test_export_dialog_builds(): def test_export_dialog_builds():
bw = FakeBalWindow() bw = FakeBalWindow()
bw.willitems = _make_willitems() bw.willitems = _make_willitems()
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin) d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
assert d.tx_strings page = d.qr_page
assert d.frames assert d.transport_qr.isChecked()
assert len(d.frames) >= 1 assert page.tx_strings
assert page.frames
assert len(page.frames) >= 1
# Frame 1 is shown. # Frame 1 is shown.
assert d.qr_view.text == d.frames[0] assert page.qr_view.text == page.frames[0]
assert "1" in d.progress_label.text() assert "1" in page.progress_label.text()
d.close()
def test_export_dialog_unified_transports():
# One window hosts the three transports as stacked, radio-selected pages.
bw = FakeBalWindow()
bw.willitems = _make_willitems()
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin)
assert d.stacked.count() == 3
assert d.transport_file.isChecked()
assert d.stacked.currentWidget() is d.file_page
d._on_mode_clicked(d.MODE_QR)
assert d.stacked.currentWidget() is d.qr_page
d._on_mode_clicked(d.MODE_AUDIO)
assert d.stacked.currentWidget() is d.audio_page
d.close()
def test_import_dialog_qr_page_has_no_audio():
# Audio lives on the import dialog's own audio page, never in the QR page.
bw = FakeBalWindow()
d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
page = d.qr_page
assert not hasattr(page, "_audio_receive")
texts = [b.text() for b in page.findChildren(dialogs.QPushButton)]
assert not any("Audio" in t for t in texts)
assert d.receive_btn is not None
d.close() d.close()
def test_export_dialog_empty_close(): def test_export_dialog_empty_close():
# An empty will shows a modal message; stub it out for the test. # An empty will shows a modal message and no widgets are built; stub the
# message out for the test.
orig = dialogs.MessageBoxMixin.show_message orig = dialogs.MessageBoxMixin.show_message
dialogs.MessageBoxMixin.show_message = lambda self, msg, icon=None: None dialogs.MessageBoxMixin.show_message = lambda self, msg, icon=None: None
try: try:
bw = FakeBalWindow() bw = FakeBalWindow()
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin) d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin)
assert not d.isVisible() assert not d.isVisible()
assert not hasattr(d, "qr_page")
d.close() d.close()
finally: finally:
dialogs.MessageBoxMixin.show_message = orig dialogs.MessageBoxMixin.show_message = orig
@@ -144,50 +191,52 @@ def test_imported_item_status_not_none():
def test_export_auto_scroll(): def test_export_auto_scroll():
bw = FakeBalWindow() bw = FakeBalWindow()
bw.willitems = _make_willitems(n=6, payload_len=400) bw.willitems = _make_willitems(n=6, payload_len=400)
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin) d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
assert d.fps_spin is not None page = d.qr_page
assert not d.auto_timer.isActive() assert page.fps_spin is not None
assert not page.auto_timer.isActive()
d.fps_spin.setValue(2) page.fps_spin.setValue(2)
d._toggle_auto() page._toggle_auto()
assert d.auto_timer.isActive() assert page.auto_timer.isActive()
assert d.auto_btn.text() == dialogs._("Stop") assert page.auto_btn.text() == dialogs._("Stop")
d._auto_step() page._auto_step()
assert d.index == 1 assert page.index == 1
d._toggle_auto() page._toggle_auto()
assert not d.auto_timer.isActive() assert not page.auto_timer.isActive()
assert d.auto_btn.text() == dialogs._("Auto") assert page.auto_btn.text() == dialogs._("Auto")
# Advancing past the last frame stops the slideshow automatically. # Advancing past the last frame stops the slideshow automatically.
d._toggle_auto() page._toggle_auto()
d.index = len(d.frames) - 1 page.index = len(page.frames) - 1
d._auto_step() page._auto_step()
assert not d.auto_timer.isActive() assert not page.auto_timer.isActive()
d.close() d.close()
def test_export_auto_scroll_loop(): def test_export_auto_scroll_loop():
bw = FakeBalWindow() bw = FakeBalWindow()
bw.willitems = _make_willitems(n=6, payload_len=400) bw.willitems = _make_willitems(n=6, payload_len=400)
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin) d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
assert d.loop_check is not None page = d.qr_page
assert page.loop_check is not None
# Loop on: reaching the last code wraps back to the first and keeps going. # Loop on: reaching the last code wraps back to the first and keeps going.
d.loop_check.setChecked(True) page.loop_check.setChecked(True)
d._toggle_auto() page._toggle_auto()
assert d.auto_timer.isActive() assert page.auto_timer.isActive()
d.index = len(d.frames) - 1 page.index = len(page.frames) - 1
d._auto_step() page._auto_step()
assert d.index == 0 assert page.index == 0
assert d.auto_timer.isActive() assert page.auto_timer.isActive()
d._toggle_auto() page._toggle_auto()
# Loop off: reaching the last code stops the slideshow. # Loop off: reaching the last code stops the slideshow.
d.loop_check.setChecked(False) page.loop_check.setChecked(False)
d._toggle_auto() page._toggle_auto()
d.index = len(d.frames) - 1 page.index = len(page.frames) - 1
d._auto_step() page._auto_step()
assert not d.auto_timer.isActive() assert not page.auto_timer.isActive()
d.close() d.close()
@@ -197,18 +246,27 @@ def test_export_filter_valid_and_valid_nc():
b = StubWillItem("B" * 120, statuses={"VALID": True}) b = StubWillItem("B" * 120, statuses={"VALID": True})
c = StubWillItem("C" * 120) c = StubWillItem("C" * 120)
bw.willitems = {"a": a, "b": b, "c": c} bw.willitems = {"a": a, "b": b, "c": c}
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin) d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
assert len(d.tx_strings) == 3 page = d.qr_page
assert d.filter_combo.count() == 3 assert d.filter_combo.count() == 3
# Whole-will default: a single JSON document carrying all selected items.
assert d.content_check.isChecked()
assert len(page.tx_strings) == 1
data = json.loads(page.tx_strings[0])
assert set(data) == {"a", "b", "c"}
# Switch to tx-only content, then exercise the filters.
d.content_check.setChecked(False)
assert len(page.tx_strings) == 3
# "Valid" filter -> only the valid items (a, b). # "Valid" filter -> only the valid items (a, b).
d._on_filter_change(1) d._on_filter_change(1)
assert sorted(d.tx_strings) == ["A" * 120, "B" * 120] assert sorted(page.tx_strings) == ["A" * 120, "B" * 120]
# "Valid NC" filter -> only the valid, not-complete item (b). # "Valid NC" filter -> only the valid, not-complete item (b).
d._on_filter_change(2) d._on_filter_change(2)
assert sorted(d.tx_strings) == ["B" * 120] assert sorted(page.tx_strings) == ["B" * 120]
assert d.qr_view.text == d.frames[0] assert page.qr_view.text == page.frames[0]
d.close() d.close()
@@ -218,116 +276,464 @@ def test_export_filter_empty_reverts():
bw.willitems = { bw.willitems = {
"a": StubWillItem("A" * 120, statuses={"VALID": True, "COMPLETE": True}) "a": StubWillItem("A" * 120, statuses={"VALID": True, "COMPLETE": True})
} }
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin) d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
messages = [] messages = []
d.show_message = lambda msg: messages.append(msg) d.show_message = lambda msg: messages.append(msg) # type: ignore[assignment]
d._on_filter_change(2) # "Valid NC" -> empty subset d._on_filter_change(2) # "Valid NC" -> empty subset
assert messages assert messages
assert d._filter_index == 0 # reverted to "All" assert d._filter_index == 0 # reverted to "All"
assert d.filter_combo.currentIndex() == 0 assert d.filter_combo.currentIndex() == 0
assert len(d.tx_strings) == 1 assert len(d.qr_page.tx_strings) == 1
d.close() d.close()
def test_export_navigation_and_chunk_change(): def test_export_navigation_and_chunk_change():
bw = FakeBalWindow() bw = FakeBalWindow()
bw.willitems = _make_willitems(n=6, payload_len=400) bw.willitems = _make_willitems(n=6, payload_len=400)
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin) d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
first_count = len(d.frames) page = d.qr_page
first_count = len(page.frames)
assert first_count > 1 # long transfer, small default chunk assert first_count > 1 # long transfer, small default chunk
assert not d.prev_btn.isEnabled() assert not page.prev_btn.isEnabled()
d._next() page._next()
assert d.index == 1 assert page.index == 1
assert d.qr_view.text == d.frames[1] assert page.qr_view.text == page.frames[1]
assert d.prev_btn.isEnabled() assert page.prev_btn.isEnabled()
d._prev() page._prev()
assert d.index == 0 assert page.index == 0
assert d.qr_view.text == d.frames[0] assert page.qr_view.text == page.frames[0]
# Switch to the largest preset: fewer, bigger frames. # Switch to the largest preset: fewer, bigger frames.
d._on_chunk_change(len(dialogs.CHUNK_PRESETS) - 1) page._on_chunk_change(len(dialogs.CHUNK_PRESETS) - 1)
assert len(d.frames) < first_count assert len(page.frames) < first_count
assert d.index == 0 assert page.index == 0
d.close() d.close()
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #
# WillQrImportDialog # WillImportDialog (QR transport via d.qr_page)
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #
def test_import_frame_flow(): def test_import_frame_flow():
bw = FakeBalWindow() bw = FakeBalWindow()
d = dialogs.WillQrImportDialog(bw, bal_plugin=bw.bal_plugin) d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
assert not d.review_btn.isEnabled() page = d.qr_page
assert not d.slot_area.isVisible() assert not page.review_btn.isEnabled()
assert not page.slot_area.isVisible()
transfer = encode_transfer(["A" * 120, "B" * 120, "C" * 120]) transfer = encode_transfer(["A" * 120, "B" * 120, "C" * 120])
frames = split_frames(transfer, 150) frames = split_frames(transfer, 150)
assert len(frames) > 1 assert len(frames) > 1
for frame in frames: for frame in frames:
d._add_frame(frame) page._add_frame(frame)
assert d.total == len(frames) assert page.total == len(frames)
assert d.review_btn.isEnabled() assert page.review_btn.isEnabled()
# isVisible() needs a shown parent; assert the widget is not hidden instead. # isVisible() needs a shown parent; assert the widget is not hidden instead.
assert not d.slot_area.isHidden() assert not page.slot_area.isHidden()
assert len(d.slot_widgets) == d.total assert len(page.slot_widgets) == page.total
assert "All" in d.status_label.text() assert "All" in page.status_label.text()
# Duplicate capture is harmless. # Duplicate capture is harmless.
d._add_frame(frames[0]) page._add_frame(frames[0])
assert len(d.frames) == d.total assert len(page.frames) == page.total
d.close() d.close()
def test_import_assembles_and_decodes(): def test_import_assembles_and_decodes():
bw = FakeBalWindow() bw = FakeBalWindow()
d = dialogs.WillQrImportDialog(bw, bal_plugin=bw.bal_plugin) d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
page = d.qr_page
captured = {} captured = {}
def fake_finish(tx_strings): def fake_finish(payload):
captured["tx_strings"] = tx_strings captured["payload"] = payload
d._finish_import = fake_finish page._finish_import = fake_finish
transfer = encode_transfer(["P" * 130, "Q" * 130]) transfer = encode_transfer(["P" * 130, "Q" * 130])
for frame in split_frames(transfer, 150): for frame in split_frames(transfer, 150):
d._add_frame(frame) page._add_frame(frame)
d._review_and_sign() page._review_and_sign()
assert captured["tx_strings"] == ["P" * 130, "Q" * 130] # The payload is rejoined into an opaque string for autodetect.
assert captured["payload"].split("\n") == ["P" * 130, "Q" * 130]
d.close() d.close()
def test_decode_will_payload_autodetect():
# Whole-will JSON is recognized as a will document.
payload = json.dumps({"item1": {"tx": "AAAA", "status": {"VALID": True}}})
kind, data = dialogs.decode_will_payload(payload)
assert kind == "will"
assert data["item1"]["tx"] == "AAAA"
# A singleton dict whose value is not an item dict falls back to txs.
kind, data = dialogs.decode_will_payload('{"foo": 1}')
assert kind == "txs"
# Comma and/or newline separated transactions.
kind, data = dialogs.decode_will_payload("AAAA,BBBB\nCCCC")
assert kind == "txs"
assert data == ["AAAA", "BBBB", "CCCC"]
# A single transaction with no separators.
kind, data = dialogs.decode_will_payload("HEXHEX")
assert kind == "txs"
assert data == ["HEXHEX"]
def test_whole_will_qr_roundtrip():
# "Whole will" produces a single JSON document that survives a full
# QR encode -> frame capture -> assemble -> decode cycle.
bw = FakeBalWindow()
items = _make_willitems(2)
bw.willitems = items
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
page = d.qr_page
assert d.content_check.isChecked()
assert len(page.tx_strings) == 1
# Rebuild the transfer from the dialog's own strings, as the importer does.
transfer = encode_transfer(page.tx_strings)
impl = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
import_page = impl.qr_page
for frame in split_frames(transfer, dialogs.CHUNK_PRESETS[0][1]):
import_page._add_frame(frame)
assert import_page.review_btn.isEnabled()
caught = {}
def fake_finish(payload):
caught["payload"] = payload
import_page._finish_import = fake_finish
import_page._review_and_sign()
kind, data = dialogs.decode_will_payload(caught["payload"])
assert kind == "will"
assert set(data) == {"item0", "item1"}
d.close()
impl.close()
def test_import_total_mismatch_resets(): def test_import_total_mismatch_resets():
bw = FakeBalWindow() bw = FakeBalWindow()
d = dialogs.WillQrImportDialog(bw, bal_plugin=bw.bal_plugin) d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
page = d.qr_page
warnings = [] warnings = []
d.show_warning = lambda msg: warnings.append(msg) page.show_warning = lambda msg: warnings.append(msg) # type: ignore[assignment]
frames_a = split_frames(encode_transfer(["A" * 120, "B" * 120]), 150) frames_a = split_frames(encode_transfer(["A" * 120, "B" * 120]), 150)
frames_b = split_frames(encode_transfer(["A" * 120, "B" * 120, "C" * 120]), 150) frames_b = split_frames(encode_transfer(["A" * 120, "B" * 120, "C" * 120]), 150)
for frame in frames_a: for frame in frames_a:
d._add_frame(frame) page._add_frame(frame)
assert d.total == len(frames_a) assert page.total == len(frames_a)
# A frame with a different total wipes the import; the first frame of # A frame with a different total wipes the import; the first frame of
# the new transfer must be scanned afresh. # the new transfer must be scanned afresh.
d._add_frame(frames_b[0]) page._add_frame(frames_b[0])
assert warnings assert warnings
assert d.total == 0 assert page.total == 0
assert not d.frames assert not page.frames
assert not d.review_btn.isEnabled() assert not page.review_btn.isEnabled()
d.close() d.close()
def test_import_manual_entry(): def test_import_manual_entry():
bw = FakeBalWindow() bw = FakeBalWindow()
d = dialogs.WillQrImportDialog(bw, bal_plugin=bw.bal_plugin) d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
page = d.qr_page
frames = split_frames(encode_transfer(["M" * 120]), 150) frames = split_frames(encode_transfer(["M" * 120]), 150)
d.manual_edit.setText(frames[0]) page.manual_edit.setText(frames[0])
d._add_from_manual() page._add_from_manual()
assert d.manual_edit.text() == "" assert page.manual_edit.text() == ""
assert d.total == 1 assert page.total == 1
assert d.review_btn.isEnabled() assert page.review_btn.isEnabled()
d.close()
# ------------------------------------------------------------------ #
# Continuous camera scan (change/detection debounce + auto-finish)
# ------------------------------------------------------------------ #
def _fresh_debounce():
return {
"last_index": None,
"last_payload": None,
"pending_index": None,
"pending_payload": None,
"pending_count": 0,
}
def test_qr_import_debounce_pending_then_accept():
s = _fresh_debounce()
# First sighting of a new identity: pending, not yet stored.
assert dialogs.qr_import_accept_frame(s, "balqr", "balqr:2", 2, 1, "PAYLOAD1") == "pending"
assert s["pending_count"] == 1
assert s["last_index"] is None
# A second stable read of the same identity: accepted.
assert dialogs.qr_import_accept_frame(s, "balqr", "balqr:2", 2, 1, "PAYLOAD1") == "accept"
assert s["last_index"] == 1
assert s["last_payload"] == "PAYLOAD1"
assert s["pending_count"] == 0
def test_qr_import_debounce_re_reading_last_is_ignored():
s = _fresh_debounce()
dialogs.qr_import_accept_frame(s, "balqr", "balqr:2", 2, 1, "P1")
dialogs.qr_import_accept_frame(s, "balqr", "balqr:2", 2, 1, "P1") # accepted
# The exporter is still showing frame 1: must be ignored, not accepted.
assert dialogs.qr_import_accept_frame(s, "balqr", "balqr:2", 2, 1, "P1") == "ignore"
assert s["last_index"] == 1
assert s["pending_count"] == 0
def test_qr_import_debounce_transition_pending_resets():
s = _fresh_debounce()
dialogs.qr_import_accept_frame(s, "balqr", "balqr:2", 2, 1, "P1")
dialogs.qr_import_accept_frame(s, "balqr", "balqr:2", 2, 1, "P1") # accept frame 1
# A new identity interrupts the pending accumulation.
assert dialogs.qr_import_accept_frame(s, "balqr", "balqr:2", 2, 2, "P2") == "pending"
assert dialogs.qr_import_accept_frame(s, "balqr", "balqr:2", 2, 2, "P2") == "accept"
# Same-index duplicate with different payload is treated as new identity.
assert dialogs.qr_import_accept_frame(s, "balqr", "balqr:2", 2, 2, "P2") == "ignore"
def test_qr_import_debounce_total_mismatch_resets():
s = _fresh_debounce()
# In-range frame is accepted even though its declared total is ignored.
assert dialogs.qr_import_accept_frame(s, "balqr", "balqr:3", 3, 2, "P2") == "pending"
assert dialogs.qr_import_accept_frame(s, "balqr", "balqr:3", 3, 2, "P2") == "accept"
# A frame that belongs to a different transfer.
assert dialogs.qr_import_accept_frame(s, "balqr", "balqr:4", 4, 2, "P2B") == "reset"
# Once the policy is rebased on the new transfer, frames resume normally
# (the widget clears the debounce while wiping the import).
s["key"] = None
assert dialogs.qr_import_accept_frame(s, "balqr", "balqr:4", 4, 2, "P2B") == "pending"
def test_qr_import_handle_scanned_text_autofinish():
bw = FakeBalWindow()
d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
page = d.qr_page
reviewed = []
page._review_and_sign = lambda: reviewed.append(True) # type: ignore[assignment]
frames = split_frames(encode_transfer(["A" * 120, "B" * 120]), 150)
assert len(frames) > 1
# The camera session is running and every frame needs two stable reads.
page._scanning = True
for frame in frames:
for _rep in range(2):
page._handle_scanned_text(frame)
assert page.total == len(frames)
assert len(page.frames) == len(frames)
assert page.review_btn.isEnabled()
# With all frames stored, the loop auto-finishes exactly once.
_app.processEvents()
assert reviewed == [True]
# The camera loop was stopped before handing over to the review step.
assert not page._scanning
assert not page._scan_timer.isActive()
d.close()
def test_qr_import_handle_scanned_text_manual_does_not_autofinish():
# Without a camera session running, extra frames never auto-proceed.
bw = FakeBalWindow()
d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
page = d.qr_page
reviewed = []
page._review_and_sign = lambda: reviewed.append(True) # type: ignore[assignment]
frames = split_frames(encode_transfer(["A" * 120, "B" * 120]), 150)
assert len(frames) > 1
assert not page._scanning
for frame in frames:
for _rep in range(2):
page._handle_scanned_text(frame)
assert len(page.frames) == len(frames)
_app.processEvents()
assert reviewed == []
d.close()
# ------------------------------------------------------------------ #
# Animated-QR formats (BC-UR v1/v2, BBQR) via the export/import pages
# ------------------------------------------------------------------ #
def test_export_format_combo_switches_codecs():
bw = FakeBalWindow()
bw.willitems = _make_willitems(n=6, payload_len=400)
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
page = d.qr_page
assert page.format == "balqr"
assert page.frames[0].startswith("BALQR1|")
assert page.format_combo.count() == 4
page._on_format_change(1) # BC-UR v1
assert page.format == "ur1"
assert page.frames[0].startswith("ur:bytes/")
assert page.index == 0
assert page.qr_view.text == page.frames[0]
page._on_format_change(2) # BC-UR v2
assert page.format == "ur2"
assert page.frames[0].startswith("ur:bytes/")
page._on_format_change(3) # BBQR
assert page.format == "bbqr"
assert page.frames[0].startswith("B$")
d.close()
def test_export_animated_format_frames_fit_budget():
bw = FakeBalWindow()
bw.willitems = _make_willitems(n=6, payload_len=400)
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
page = d.qr_page
for index in range(1, 4):
page._on_format_change(index)
for frame in page.frames:
assert len(frame) <= dialogs.CHUNK_PRESETS[0][1]
d.close()
def _import_roundtrip_fmt(fmt_index):
bw = FakeBalWindow()
bw.willitems = _make_willitems(2)
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
page = d.qr_page
page._on_format_change(fmt_index)
frames = list(page.frames)
assert frames
d.close()
impl = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
import_page = impl.qr_page
caught = {}
import_page._finish_import = lambda payload: caught.__setitem__("payload", payload)
for frame in frames:
import_page._add_frame(frame)
assert import_page.review_btn.isEnabled()
import_page._review_and_sign()
kind, data = dialogs.decode_will_payload(caught["payload"])
assert kind == "will"
assert set(data) == {"item0", "item1"}
impl.close()
def test_import_ur1_roundtrip():
_import_roundtrip_fmt(1)
def test_import_ur2_roundtrip():
_import_roundtrip_fmt(2)
def test_import_bbqr_roundtrip():
_import_roundtrip_fmt(3)
def test_import_animated_scan_debounce_autofinish():
bw = FakeBalWindow()
bw.willitems = _make_willitems(2)
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
page = d.qr_page
page._on_format_change(2) # BC-UR v2 fountain
frames = list(page.frames)
d.close()
impl = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
import_page = impl.qr_page
reviewed = []
import_page._review_and_sign = lambda: reviewed.append(True) # type: ignore[assignment]
import_page._scanning = True
for frame in frames:
for _rep in range(2):
import_page._handle_scanned_text(frame)
assert import_page.review_btn.isEnabled()
# The fountain transfer's part count is ``len(frames) // 2`` (pure + one
# redundant mixed wave).
assert import_page.total == len(frames) // 2
assert len(import_page.frames) >= import_page.total
assert import_page.review_btn.isEnabled()
_app.processEvents()
assert reviewed == [True]
assert not import_page._scanning
impl.close()
def test_import_garbage_scan_is_ignored():
bw = FakeBalWindow()
d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
page = d.qr_page
for garbage in ("hello world", "12345", "B$ZZ", ""):
page._handle_scanned_text(garbage)
assert not page.frames
assert page.total == 0
assert not page.review_btn.isEnabled()
d.close()
def test_import_different_animated_transfer_resets():
bw = FakeBalWindow()
d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
page = d.qr_page
warnings = []
page.show_warning = lambda msg: warnings.append(msg) # type: ignore[assignment]
frames_a = split_frames(encode_transfer(["A" * 120, "B" * 120]), 150)
frames_b = split_frames(encode_transfer(["A" * 120, "B" * 120, "C" * 120]), 150)
for frame in frames_a:
page._add_frame(frame)
assert page.total == len(frames_a)
assert not warnings
page._add_frame(frames_b[0])
assert warnings
assert page.total == 0
assert not page.frames
assert not page.review_btn.isEnabled()
d.close()
def test_import_start_stop_scan_signal_wiring():
"""Regression: _start_scan/_stop_scan must use the QVideoSink signal
videoFrameChanged, not the videoFrame frame getter.
On PyQt6, ``QVideoSink.videoFrame`` is a method (the frame getter), so
``.videoFrame.connect(...)`` raises AttributeError. This test drives the
real sink life-cycle with a mocked camera and asserts the scan session
starts/ends cleanly with no error.
"""
from PyQt6.QtMultimedia import QCamera, QMediaCaptureSession, QMediaDevices
bw = FakeBalWindow()
d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
page = d.qr_page
errors = []
page.show_error = lambda msg: errors.append(msg) # type: ignore[assignment]
fake_device = MagicMock()
fake_device.isNull.return_value = False
with (
patch.object(QMediaDevices, "defaultVideoInput", return_value=fake_device),
# Mock camera + capture session; the QVideoSink stays real so the
# videoFrameChanged connect/disconnect wiring is exercised for real.
patch.object(QCamera, "__new__", return_value=MagicMock()),
patch.object(QMediaCaptureSession, "__new__", return_value=MagicMock()),
):
page._start_scan()
assert page._scanning is True
assert not errors
page._stop_scan()
assert page._scanning is False
assert page._camera is None
assert page._video_sink is None
assert not errors
d.close() d.close()

View File

@@ -28,7 +28,6 @@ Run:
python3 tests/test_heir_relative_anchor.py python3 tests/test_heir_relative_anchor.py
""" """
import copy
import json import json
import os import os
import sys import sys
@@ -40,6 +39,7 @@ from electrum import constants # noqa: E402 (path insert above)
constants.net = constants.BitcoinRegtest constants.net = constants.BitcoinRegtest
from bal.core.checkalive import resolve_date_to_check # noqa: E402 from bal.core.checkalive import resolve_date_to_check # noqa: E402
from bal.core.util import copy_structure # noqa: E402
from bal.core.will import ( # noqa: E402 from bal.core.will import ( # noqa: E402
HeirNotFoundException, HeirNotFoundException,
NoHeirsException, NoHeirsException,
@@ -71,7 +71,7 @@ def _make_will_item(heirs, tx_locktime, status_complete=False):
is forced to ``tx_locktime`` (the value frozen in the signed Bitcoin tx).""" is forced to ``tx_locktime`` (the value frozen in the signed Bitcoin tx)."""
d = { d = {
"tx": _VALID_TX_HEX, "tx": _VALID_TX_HEX,
"heirs": copy.deepcopy(heirs), "heirs": copy_structure(heirs),
"willexecutor": None, "willexecutor": None,
"status": "", "status": "",
"description": "", "description": "",
@@ -80,7 +80,7 @@ def _make_will_item(heirs, tx_locktime, status_complete=False):
"baltx_fees": 1, "baltx_fees": 1,
} }
item = WillItem(d, _id="willid_1") item = WillItem(d, _id="willid_1")
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT) item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
item.tx.locktime = tx_locktime item.tx.locktime = tx_locktime
if status_complete: if status_complete:
item.set_status("COMPLETE", True) item.set_status("COMPLETE", True)
@@ -111,7 +111,7 @@ def test_unchanged_relative_recipe_signed_is_coherent():
read as a postpone just because the clock has advanced past build day.""" read as a postpone just because the clock has advanced past build day."""
heirs = {"alice": ["addr_alice", 5000, "1y"]} heirs = {"alice": ["addr_alice", 5000, "1y"]}
outcome = _run_heir_check( outcome = _run_heir_check(
copy.deepcopy(heirs), copy.deepcopy(heirs), _FROZEN, status_complete=True copy_structure(heirs), copy_structure(heirs), _FROZEN, status_complete=True
) )
assert outcome.startswith("coherent"), outcome assert outcome.startswith("coherent"), outcome
@@ -119,7 +119,7 @@ def test_unchanged_relative_recipe_signed_is_coherent():
def test_unchanged_relative_recipe_unsigned_is_coherent(): def test_unchanged_relative_recipe_unsigned_is_coherent():
heirs = {"alice": ["addr_alice", 5000, "1y"]} heirs = {"alice": ["addr_alice", 5000, "1y"]}
outcome = _run_heir_check( outcome = _run_heir_check(
copy.deepcopy(heirs), copy.deepcopy(heirs), _FROZEN, status_complete=False copy_structure(heirs), copy_structure(heirs), _FROZEN, status_complete=False
) )
assert outcome.startswith("coherent"), outcome assert outcome.startswith("coherent"), outcome
@@ -145,7 +145,7 @@ def test_relative_recipe_shortened_on_signed_is_rebuild():
def test_unchanged_absolute_recipe_is_coherent(): def test_unchanged_absolute_recipe_is_coherent():
built = {"alice": ["addr_alice", 5000, str(_FROZEN)]} built = {"alice": ["addr_alice", 5000, str(_FROZEN)]}
outcome = _run_heir_check( outcome = _run_heir_check(
copy.deepcopy(built), copy.deepcopy(built), _FROZEN, status_complete=True copy_structure(built), copy_structure(built), _FROZEN, status_complete=True
) )
assert outcome.startswith("coherent"), outcome assert outcome.startswith("coherent"), outcome
@@ -176,6 +176,7 @@ def test_karen7_frozen_delivery_not_expired():
valid_wid = "28b64bfd83878d15c668473aa695a2b9bc23196bc61ab3149e8f33241826978d" valid_wid = "28b64bfd83878d15c668473aa695a2b9bc23196bc61ab3149e8f33241826978d"
wi = WillItem(data["will"][valid_wid], _id=valid_wid) wi = WillItem(data["will"][valid_wid], _id=valid_wid)
built_locktime = Will.get_min_locktime({valid_wid: wi}) built_locktime = Will.get_min_locktime({valid_wid: wi})
assert built_locktime is not None
assert built_locktime == int(wi.tx.locktime) assert built_locktime == int(wi.tx.locktime)
date_to_check = resolve_date_to_check( date_to_check = resolve_date_to_check(
@@ -195,7 +196,6 @@ def test_karen7_unchanged_heirs_are_coherent():
signed tx: the plugin must NOT ask to invalidate the will.""" signed tx: the plugin must NOT ask to invalidate the will."""
data = _load_karen7() data = _load_karen7()
valid_wid = "28b64bfd83878d15c668473aa695a2b9bc23196bc61ab3149e8f33241826978d" valid_wid = "28b64bfd83878d15c668473aa695a2b9bc23196bc61ab3149e8f33241826978d"
wi = WillItem(data["will"][valid_wid], _id=valid_wid)
# Use _FROZEN (a UTC-midnight value) so the check is compatible with # Use _FROZEN (a UTC-midnight value) so the check is compatible with
# the UTC anchoring code. # the UTC anchoring code.
frozen_locktime = _FROZEN frozen_locktime = _FROZEN

View File

@@ -127,6 +127,11 @@ def test_sign_transactions_external_only():
wallet=FakeWallet(), wallet=FakeWallet(),
waiting_dialog=SimpleNamespace(update=lambda msg: None), waiting_dialog=SimpleNamespace(update=lambda msg: None),
) )
# sign_transactions dispatches to self._prepare_and_sign_tx; bind the real
# implementation onto the fake so the external-sign run actually executes.
fake._prepare_and_sign_tx = MethodType(
window_mod.BalWindow._prepare_and_sign_tx, fake
)
result = window_mod.BalWindow.sign_transactions(fake, None, will=imported) result = window_mod.BalWindow.sign_transactions(fake, None, will=imported)

View File

@@ -27,7 +27,6 @@ Run::
python3 -m pytest tests/test_no_willexecutor_karen7.py -v -s python3 -m pytest tests/test_no_willexecutor_karen7.py -v -s
""" """
import copy
import json import json
import logging import logging
import os import os
@@ -49,6 +48,7 @@ from electrum.transaction import PartialTxInput, TxOutpoint
from electrum.util import bfh from electrum.util import bfh
from bal.core.heirs import Heirs from bal.core.heirs import Heirs
from bal.core.util import copy_structure
from bal.core.will import ( from bal.core.will import (
NotCompleteWillException, NotCompleteWillException,
NoWillExecutorNotPresent, NoWillExecutorNotPresent,
@@ -278,11 +278,11 @@ class FakeBalWindow:
tx["my_locktime"] = txs[txid].my_locktime tx["my_locktime"] = txs[txid].my_locktime
tx["heirsvalue"] = txs[txid].heirsvalue tx["heirsvalue"] = txs[txid].heirsvalue
tx["description"] = txs[txid].description tx["description"] = txs[txid].description
tx["willexecutor"] = copy.deepcopy(txs[txid].willexecutor) tx["willexecutor"] = copy_structure(txs[txid].willexecutor)
tx["status"] = "New" tx["status"] = "New"
tx["baltx_fees"] = txs[txid].tx_fees tx["baltx_fees"] = txs[txid].tx_fees
tx["time"] = creation_time tx["time"] = creation_time
tx["heirs"] = copy.deepcopy(txs[txid].heirs) tx["heirs"] = copy_structure(txs[txid].heirs)
tx["txchildren"] = [] tx["txchildren"] = []
will[txid] = WillItem(tx, _id=txid, wallet=self.wallet) will[txid] = WillItem(tx, _id=txid, wallet=self.wallet)
self.update_will(will) self.update_will(will)

View File

@@ -7,7 +7,6 @@ but without requiring a full Qt event loop.
""" """
import contextlib import contextlib
import copy
import json import json
import os import os
import sys import sys
@@ -24,6 +23,7 @@ if os.path.isdir(ELECTRUM_DIR):
from bal.core.heirs import Heirs from bal.core.heirs import Heirs
from bal.core.plugin_base import BalPlugin, BalTimestamp from bal.core.plugin_base import BalPlugin, BalTimestamp
from bal.core.util import copy_structure
from bal.core.will import ( from bal.core.will import (
NoHeirsException, NoHeirsException,
NotCompleteWillException, NotCompleteWillException,
@@ -145,11 +145,11 @@ class FakeBalWindow:
tx["my_locktime"] = txs[txid].my_locktime tx["my_locktime"] = txs[txid].my_locktime
tx["heirsvalue"] = txs[txid].heirsvalue tx["heirsvalue"] = txs[txid].heirsvalue
tx["description"] = txs[txid].description tx["description"] = txs[txid].description
tx["willexecutor"] = copy.deepcopy(txs[txid].willexecutor) tx["willexecutor"] = copy_structure(txs[txid].willexecutor)
tx["status"] = "New" tx["status"] = "New"
tx["baltx_fees"] = txs[txid].tx_fees tx["baltx_fees"] = txs[txid].tx_fees
tx["time"] = creation_time tx["time"] = creation_time
tx["heirs"] = copy.deepcopy(txs[txid].heirs) tx["heirs"] = copy_structure(txs[txid].heirs)
tx["txchildren"] = [] tx["txchildren"] = []
will[txid] = WillItem(tx, _id=txid, wallet=self.wallet) will[txid] = WillItem(tx, _id=txid, wallet=self.wallet)
Will.update_will(self.willitems, will) Will.update_will(self.willitems, will)