core+gui: chunked multi-QR will transfer (export/import + review/sign wizard, audio channel)

BALQR-framed chunk scheduler (bal/core/qrtransfer.py) with zlib compression,
four chunk presets and a QR_CHUNK_SIZE setting (row 16).
Export/import dialogs (WillQrExportDialog/WillQrImportDialog), per-tx
review-and-sign wizard, export filters (All/Valid/Valid-NC) and an Auto
slideshow with speed + loop for the QR codes; optional audio_modem channel
with a local receive mirror. _prepare_and_sign_tx refactor (no behaviour
change); WillItem.status defaults to '' instead of None (import crash fix);
invalidate_will guard for a missing date_to_check. Docs: PLAN_QR_TRANSFER,
QML_PLAN, README, HANDOFF, CHANGELOG entry 56, AUDIO_MODEM_DEBIAN.
This commit is contained in:
2026-08-28 16:28:37 -04:00
parent 0bacbf1997
commit d288b553ef
16 changed files with 3106 additions and 64 deletions

View File

@@ -0,0 +1,327 @@
"""
Tests for ``bal.core.qrtransfer``.
Covers the BALQR frame encoding used for will transfer via QR codes /
audio modem: encoding, framing, reassembly, malformed input and the preset
list (optionally cross-checked against the ``qrcode`` library's EC-M
capacity when it is installed).
Run:
source electrum/env/bin/activate
python3 tests/test_core_qr_transfer.py
"""
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
from bal.core.qrtransfer import (
CHUNK_PRESETS,
MIN_CHUNK_SIZE,
InconsistentTotalError,
MissingFramesError,
QrTransferError,
assemble,
decode_transfer,
encode_transfer,
parse_frame,
preset_index_for_chunk_size,
split_frames,
)
def _frames(tx_strings, chunk_size, compress=False):
"""Split a payload and return (payload, total, {index: payload})."""
payload = encode_transfer(tx_strings, compress=compress)
parsed = {}
total = None
for frame in split_frames(payload, chunk_size, compressed=compress):
t, index, _compressed, p = parse_frame(frame)
if total is not None:
assert total == t
total = t
parsed[index] = p
assert total is not None
return payload, total, parsed
# --------------------------------------------------------------------------- #
# Round trips
# --------------------------------------------------------------------------- #
def test_encode_decode_plain():
tx_strings = ["00" * 32, "aa" * 40, "ff" * 50]
payload = encode_transfer(tx_strings, compress=False)
assert decode_transfer(payload, compressed=False) == tx_strings
def test_encode_decode_compressed():
tx_strings = ["00" * 32, "aa" * 40, "ff" * 50]
payload = encode_transfer(tx_strings, compress=True)
assert decode_transfer(payload, compressed=True) == tx_strings
def test_empty_list_roundtrip():
assert decode_transfer(encode_transfer([]), compressed=False) == []
# --------------------------------------------------------------------------- #
# Framing
# --------------------------------------------------------------------------- #
def test_single_frame():
tx_strings = ["11" * 10]
payload = encode_transfer(tx_strings)
frames = split_frames(payload, 150)
assert len(frames) == 1
total, index, compressed, p = parse_frame(frames[0])
assert (total, index, compressed) == (1, 1, False)
assert p == payload
def test_multiple_frames_reassemble():
tx_strings = ["ab" * 100, "cd" * 100] # 600 chars -> multiple frames
_payload, total, parsed = _frames(tx_strings, CHUNK_PRESETS[0][1])
assert total > 1
decoded = decode_transfer(assemble(parsed, total), compressed=False)
assert decoded == tx_strings
def test_size_greater_than_payload():
tx_strings = ["12" * 5]
payload = encode_transfer(tx_strings)
frames = split_frames(payload, 1800)
assert len(frames) == 1
_t, _i, _c, p = parse_frame(frames[0])
assert p == payload
def test_exact_single_frame_boundary():
# A 138-byte payload exactly fills the 150-byte preset budget (the 12-char
# empty-flags header plus payload), so the encoded frame is exactly 150.
tx_strings = ["a" * 138]
payload = encode_transfer(tx_strings)
frames = split_frames(payload, 150)
assert len(frames) == 1
assert len(frames[0]) == 150
_t, _i, _c, p = parse_frame(frames[0])
assert p == payload
def test_frames_fit_chunk_size():
tx_strings = ["".join("{:02x}".format(i) * 2) for i in range(300)]
payload = encode_transfer(tx_strings)
for _label, size in CHUNK_PRESETS:
for frame in split_frames(payload, size):
assert len(frame) <= size, (size, len(frame))
def test_single_tx_larger_than_chunk():
# A huge serialized tx must be split over several frames and reassemble
# exactly (positional slicing is safe for hex/base64 text).
tx_strings = ["7b" * 1000] # 2000 chars
payload = encode_transfer(tx_strings)
frames = split_frames(payload, 150)
assert len(frames) > 1
parsed = {parse_frame(f)[1]: parse_frame(f)[3] for f in frames}
total = parse_frame(frames[0])[0]
assert assemble(parsed, total) == payload
def test_compressed_frames_carry_flag():
tx_strings = ["ab" * 40]
frames = split_frames(encode_transfer(tx_strings, compress=True), 150, compressed=True)
for frame in frames:
_t, _i, compressed, _p = parse_frame(frame)
assert compressed is True
# Plain frames do not.
frames_plain = split_frames(encode_transfer(tx_strings), 150)
_t, _i, compressed, _p = parse_frame(frames_plain[0])
assert compressed is False
def test_compressed_roundtrip_through_frames():
tx_strings = ["ab" * 50, "cd" * 50, "12" * 60]
payload = encode_transfer(tx_strings, compress=True)
parsed = {}
total = None
for frame in split_frames(payload, 400, compressed=True):
t, index, _c, p = parse_frame(frame)
total = t
parsed[index] = p
assert total is not None
decoded = decode_transfer(assemble(parsed, total), compressed=True)
assert decoded == tx_strings
# --------------------------------------------------------------------------- #
# Malformed input
# --------------------------------------------------------------------------- #
def test_parse_bad_magic_and_version():
for frame in (
"BALQR|1|1||a", # missing version
"BALQR2|1|1||a", # unknown version
"XXXXX1|1|1||a", # unknown magic
):
try:
parse_frame(frame)
except QrTransferError:
pass
else:
raise AssertionError("expected QrTransferError for: {}".format(frame))
def test_parse_bad_arity():
for frame in ("BALQR1", "BALQR1|1|1|"):
try:
parse_frame(frame)
except QrTransferError:
pass
else:
raise AssertionError("expected QrTransferError for: {}".format(frame))
def test_parse_pipe_in_payload_is_folded():
# maxsplit keeps the tail (including any inner '|') in the payload part.
frame = "BALQR1|1|1||a|b|c"
total, index, compressed, payload = parse_frame(frame)
assert (total, index, compressed) == (1, 1, False)
assert payload == "a|b|c"
def test_parse_bad_numbers():
for frame in (
"BALQR1|x|1||a",
"BALQR1|1|y||a",
"BALQR1|0|1||a",
"BALQR1|1|0||a",
"BALQR1|1|2||a", # index beyond total
"BALQR1|-1|1||a",
):
try:
parse_frame(frame)
except QrTransferError:
pass
else:
raise AssertionError("expected QrTransferError for: {}".format(frame))
def test_parse_bad_flags():
try:
parse_frame("BALQR1|1|1|Q|payload")
except QrTransferError:
pass
else:
raise AssertionError("expected QrTransferError for unknown flags")
def test_assemble_missing_frames():
try:
assemble({1: "a", 3: "c"}, total=3)
except MissingFramesError as e:
assert e.missing == [2]
else:
raise AssertionError("expected MissingFramesError")
def test_assemble_index_beyond_total():
try:
assemble({1: "a", 2: "b"}, total=1)
except InconsistentTotalError:
pass
else:
raise AssertionError("expected InconsistentTotalError")
def test_assemble_order_and_total_validation():
assert assemble({1: "a", 2: "b"}, total=2) == "ab"
try:
assemble({}, total=0)
except QrTransferError:
pass
else:
raise AssertionError("expected QrTransferError")
# --------------------------------------------------------------------------- #
# Constants / presets
# --------------------------------------------------------------------------- #
def test_preset_count_and_order():
assert len(CHUNK_PRESETS) == 4
budgets = [budget for _label, budget in CHUNK_PRESETS]
assert budgets == sorted(budgets)
def test_preset_index_for_chunk_size():
for index, (_label, budget) in enumerate(CHUNK_PRESETS):
assert preset_index_for_chunk_size(budget) == index
assert preset_index_for_chunk_size(150) == 0
assert preset_index_for_chunk_size(1800) == 3
def test_min_chunk_size_guard():
try:
split_frames("x" * 10, MIN_CHUNK_SIZE - 1)
except QrTransferError:
pass
else:
raise AssertionError("expected QrTransferError for tiny chunk size")
def test_split_frame_headers_consistent():
tx_strings = ["ab" * 80]
payload = encode_transfer(tx_strings)
frames = split_frames(payload, 150)
totals = {parse_frame(frame)[0] for frame in frames}
assert len(totals) == 1
assert totals.pop() == len(frames)
# --------------------------------------------------------------------------- #
# Optional: cross-check presets against the qrcode library (EC level M)
# --------------------------------------------------------------------------- #
def test_presets_fit_qrcode_ec_m():
"""Every preset budget must render inside a QR at EC level M."""
try:
import qrcode
from qrcode.constants import ERROR_CORRECT_M
except ImportError:
print("qrcode not installed - skipping capacity check")
return
for _label, size in CHUNK_PRESETS:
# Worst-case frame: header with the largest plausible total/index plus
# a full payload of the preset budget.
frame = "BALQR1|9999|9999|Z|" + "a" * (size - 14)
qr = qrcode.QRCode(error_correction=ERROR_CORRECT_M, border=2)
qr.add_data(frame)
qr.get_matrix() # raises DataOverflowError if it does not fit
# --------------------------------------------------------------------------- #
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

@@ -0,0 +1,343 @@
"""
Tests for the QR / audio will-transfer dialogs (``bal.gui.qt.dialogs``).
Covers WillQrExportDialog (build, frame navigation, chunk-size change) and
WillQrImportDialog (frame capture, slot grid, complete-review enabling, total
mismatch reset, frame assembly -> decode). The wizard and the camera/audio
paths need a live wallet/hardware and are exercised only through the shared
frame-assembly path here.
Run:
QT_QPA_PLATFORM=offscreen python3 tests/test_gui_qr_transfer.py
"""
import base64
import sys
from unittest.mock import patch
sys.path.insert(0, __file__.rsplit("/", 2)[0])
from electrum.transaction import Transaction
from PyQt6.QtWidgets import QApplication, QMainWindow
import bal.gui.qt.dialogs as dialogs
from bal.core.qrtransfer import encode_transfer, split_frames
from bal.core.will import WillItem
_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),
# reused for the WillItem status regression test.
_VALID_TX_HEX = (
"01000000012a5c9a94fcde98f5581cd00162c60a13936ceb75389ea65b"
"f38633b424eb4031000000006c493046022100a82bbc57a0136751e543"
"3f41cf000b3f1a99c6744775e76ec764fb78c54ee100022100f9e80b7d"
"e89de861dc6fb0c1429d5da72c2b6b2ee2406bc9bfb1beedd729d98501"
"2102e61d176da16edd1d258a200ad9759ef63adf8e14cd97f53227bae3"
"5cdb84d2f6ffffffff0140420f00000000001976a914230ac37834073a"
"42146f11ef8414ae929feaafc388ac00000000"
)
class FakePlugin:
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 FakeBalWindow:
"""Duck-typed stand-in for BalWindow (dialog layer only)."""
def __init__(self):
self.window = FakeWindow()
self.bal_plugin = FakePlugin()
self.willitems = {}
def get_audio_modem_plugin(self):
return None
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 _make_willitems(n=3, payload_len=120):
return {
"item{}".format(i): StubWillItem("T{}".format(i) * payload_len)
for i in range(n)
}
# ------------------------------------------------------------------ #
# WillQrExportDialog
# ------------------------------------------------------------------ #
def test_export_dialog_builds():
bw = FakeBalWindow()
bw.willitems = _make_willitems()
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin)
assert d.tx_strings
assert d.frames
assert len(d.frames) >= 1
# Frame 1 is shown.
assert d.qr_view.text == d.frames[0]
assert "1" in d.progress_label.text()
d.close()
def test_export_dialog_empty_close():
# An empty will shows a modal message; stub it out for the test.
orig = dialogs.MessageBoxMixin.show_message
dialogs.MessageBoxMixin.show_message = lambda self, msg, icon=None: None
try:
bw = FakeBalWindow()
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin)
assert not d.isVisible()
d.close()
finally:
dialogs.MessageBoxMixin.show_message = orig
def test_imported_item_status_not_none():
# Regression: a WillItem built from a bare {"tx": hex} had a None status,
# so set_status (e.g. IMPORTED / INVALIDATED from the import validity
# pass) crashed with "unsupported operand type(s) for +=: 'NoneType' and
# 'str'".
with patch.object(Transaction, "add_info_from_wallet"):
wi = WillItem({"tx": _VALID_TX_HEX}, wallet=None)
assert wi.status == ""
assert wi.set_status("IMPORTED", True) is True
assert wi.set_status("INVALIDATED", True) is True
assert "Imported" in wi.status and "Invalidated" in wi.status
def test_export_auto_scroll():
bw = FakeBalWindow()
bw.willitems = _make_willitems(n=6, payload_len=400)
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin)
assert d.fps_spin is not None
assert not d.auto_timer.isActive()
d.fps_spin.setValue(2)
d._toggle_auto()
assert d.auto_timer.isActive()
assert d.auto_btn.text() == dialogs._("Stop")
d._auto_step()
assert d.index == 1
d._toggle_auto()
assert not d.auto_timer.isActive()
assert d.auto_btn.text() == dialogs._("Auto")
# Advancing past the last frame stops the slideshow automatically.
d._toggle_auto()
d.index = len(d.frames) - 1
d._auto_step()
assert not d.auto_timer.isActive()
d.close()
def test_export_auto_scroll_loop():
bw = FakeBalWindow()
bw.willitems = _make_willitems(n=6, payload_len=400)
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin)
assert d.loop_check is not None
# Loop on: reaching the last code wraps back to the first and keeps going.
d.loop_check.setChecked(True)
d._toggle_auto()
assert d.auto_timer.isActive()
d.index = len(d.frames) - 1
d._auto_step()
assert d.index == 0
assert d.auto_timer.isActive()
d._toggle_auto()
# Loop off: reaching the last code stops the slideshow.
d.loop_check.setChecked(False)
d._toggle_auto()
d.index = len(d.frames) - 1
d._auto_step()
assert not d.auto_timer.isActive()
d.close()
def test_export_filter_valid_and_valid_nc():
bw = FakeBalWindow()
a = StubWillItem("A" * 120, statuses={"VALID": True, "COMPLETE": True})
b = StubWillItem("B" * 120, statuses={"VALID": True})
c = StubWillItem("C" * 120)
bw.willitems = {"a": a, "b": b, "c": c}
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin)
assert len(d.tx_strings) == 3
assert d.filter_combo.count() == 3
# "Valid" filter -> only the valid items (a, b).
d._on_filter_change(1)
assert sorted(d.tx_strings) == ["A" * 120, "B" * 120]
# "Valid NC" filter -> only the valid, not-complete item (b).
d._on_filter_change(2)
assert sorted(d.tx_strings) == ["B" * 120]
assert d.qr_view.text == d.frames[0]
d.close()
def test_export_filter_empty_reverts():
bw = FakeBalWindow()
# Only a COMPLETE valid item: "Valid NC" selects nothing -> revert.
bw.willitems = {
"a": StubWillItem("A" * 120, statuses={"VALID": True, "COMPLETE": True})
}
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin)
messages = []
d.show_message = lambda msg: messages.append(msg)
d._on_filter_change(2) # "Valid NC" -> empty subset
assert messages
assert d._filter_index == 0 # reverted to "All"
assert d.filter_combo.currentIndex() == 0
assert len(d.tx_strings) == 1
d.close()
def test_export_navigation_and_chunk_change():
bw = FakeBalWindow()
bw.willitems = _make_willitems(n=6, payload_len=400)
d = dialogs.WillQrExportDialog(bw, bal_plugin=bw.bal_plugin)
first_count = len(d.frames)
assert first_count > 1 # long transfer, small default chunk
assert not d.prev_btn.isEnabled()
d._next()
assert d.index == 1
assert d.qr_view.text == d.frames[1]
assert d.prev_btn.isEnabled()
d._prev()
assert d.index == 0
assert d.qr_view.text == d.frames[0]
# Switch to the largest preset: fewer, bigger frames.
d._on_chunk_change(len(dialogs.CHUNK_PRESETS) - 1)
assert len(d.frames) < first_count
assert d.index == 0
d.close()
# ------------------------------------------------------------------ #
# WillQrImportDialog
# ------------------------------------------------------------------ #
def test_import_frame_flow():
bw = FakeBalWindow()
d = dialogs.WillQrImportDialog(bw, bal_plugin=bw.bal_plugin)
assert not d.review_btn.isEnabled()
assert not d.slot_area.isVisible()
transfer = encode_transfer(["A" * 120, "B" * 120, "C" * 120])
frames = split_frames(transfer, 150)
assert len(frames) > 1
for frame in frames:
d._add_frame(frame)
assert d.total == len(frames)
assert d.review_btn.isEnabled()
# isVisible() needs a shown parent; assert the widget is not hidden instead.
assert not d.slot_area.isHidden()
assert len(d.slot_widgets) == d.total
assert "All" in d.status_label.text()
# Duplicate capture is harmless.
d._add_frame(frames[0])
assert len(d.frames) == d.total
d.close()
def test_import_assembles_and_decodes():
bw = FakeBalWindow()
d = dialogs.WillQrImportDialog(bw, bal_plugin=bw.bal_plugin)
captured = {}
def fake_finish(tx_strings):
captured["tx_strings"] = tx_strings
d._finish_import = fake_finish
transfer = encode_transfer(["P" * 130, "Q" * 130])
for frame in split_frames(transfer, 150):
d._add_frame(frame)
d._review_and_sign()
assert captured["tx_strings"] == ["P" * 130, "Q" * 130]
d.close()
def test_import_total_mismatch_resets():
bw = FakeBalWindow()
d = dialogs.WillQrImportDialog(bw, bal_plugin=bw.bal_plugin)
warnings = []
d.show_warning = lambda msg: warnings.append(msg)
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:
d._add_frame(frame)
assert d.total == len(frames_a)
# A frame with a different total wipes the import; the first frame of
# the new transfer must be scanned afresh.
d._add_frame(frames_b[0])
assert warnings
assert d.total == 0
assert not d.frames
assert not d.review_btn.isEnabled()
d.close()
def test_import_manual_entry():
bw = FakeBalWindow()
d = dialogs.WillQrImportDialog(bw, bal_plugin=bw.bal_plugin)
frames = split_frames(encode_transfer(["M" * 120]), 150)
d.manual_edit.setText(frames[0])
d._add_from_manual()
assert d.manual_edit.text() == ""
assert d.total == 1
assert d.review_btn.isEnabled()
d.close()
# ------------------------------------------------------------------ #
# Main
# ------------------------------------------------------------------ #
if __name__ == "__main__":
for name in sorted(dir()):
if name.startswith("test_"):
globals()[name]()
print(f" [OK] {name}")
print("[OK] All QR transfer GUI tests passed")