core+gui+cli: animated QR will transfer (balqr/UR1/UR2/BBQR codecs, audio channel, export/import wizard)
This commit is contained in:
@@ -21,12 +21,12 @@ Run:
|
||||
QT_QPA_PLATFORM=offscreen PYTHONPATH=electrum-src python3 tests/sim_update_flows.py
|
||||
"""
|
||||
|
||||
import copy
|
||||
import os
|
||||
import sys
|
||||
|
||||
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 (
|
||||
HeirNotFoundException,
|
||||
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)."""
|
||||
d = {
|
||||
"tx": _VALID_TX_HEX,
|
||||
"heirs": copy.deepcopy(heirs),
|
||||
"heirs": copy_structure(heirs),
|
||||
"willexecutor": None,
|
||||
"status": "",
|
||||
"description": "",
|
||||
@@ -67,7 +67,7 @@ def _make_will_item(heirs, tx_locktime, status_complete=False):
|
||||
"baltx_fees": TX_FEES,
|
||||
}
|
||||
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.
|
||||
item.tx.locktime = tx_locktime
|
||||
if status_complete:
|
||||
@@ -118,7 +118,7 @@ def main():
|
||||
# Scenario 0: nothing changed -> should be coherent.
|
||||
heirs = {"alice": ["addr_alice", 5000, same_lt]}
|
||||
_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)
|
||||
|
||||
# Scenario 1: delivery date moved forward (postpone), will NOT yet signed.
|
||||
|
||||
@@ -27,7 +27,6 @@ Run:
|
||||
tests/test_anticipate_manual_locktime.py -q
|
||||
"""
|
||||
|
||||
import copy
|
||||
import os
|
||||
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]
|
||||
|
||||
from bal.core.util import copy_structure # noqa: E402
|
||||
from bal.core.will import ( # noqa: E402
|
||||
NotCompleteWillException,
|
||||
Will,
|
||||
@@ -70,7 +70,7 @@ def _make_will_item(heirs, tx_locktime, status_complete=False):
|
||||
"""
|
||||
d = {
|
||||
"tx": _VALID_TX_HEX,
|
||||
"heirs": copy.deepcopy(heirs),
|
||||
"heirs": copy_structure(heirs),
|
||||
"willexecutor": None,
|
||||
"status": "",
|
||||
"description": "",
|
||||
@@ -79,7 +79,7 @@ def _make_will_item(heirs, tx_locktime, status_complete=False):
|
||||
"baltx_fees": TX_FEES,
|
||||
}
|
||||
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
|
||||
if status_complete:
|
||||
item.set_status("COMPLETE", True)
|
||||
|
||||
478
tests/test_core_animated_qr.py
Normal file
478
tests/test_core_animated_qr.py
Normal 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")
|
||||
@@ -14,7 +14,7 @@ import time
|
||||
|
||||
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
|
||||
|
||||
@@ -74,7 +74,7 @@ def test_bt_to_date_absolute():
|
||||
|
||||
|
||||
def test_bt_to_date_relative():
|
||||
now = datetime.now()
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
# relative days from now
|
||||
bt = BalTimestamp("7d")
|
||||
@@ -86,8 +86,8 @@ def test_bt_to_date_relative():
|
||||
d_rev = bt.to_date(reverse=True)
|
||||
assert d_rev < now
|
||||
|
||||
# from explicit datetime
|
||||
base = datetime(2025, 6, 1, 12, 0, 0)
|
||||
# from explicit datetime (UTC, so the naive-timestamp roundtrip below is stable)
|
||||
base = datetime(2025, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
d = bt.to_date(from_date=base)
|
||||
expected = (base + timedelta(days=7)).replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
assert d == expected
|
||||
@@ -101,7 +101,7 @@ def test_bt_to_date_relative():
|
||||
def test_bt_to_date_years():
|
||||
bt = BalTimestamp("1y")
|
||||
d = bt.to_date()
|
||||
assert d > datetime.now()
|
||||
assert d > datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def test_bt_to_date_overflow():
|
||||
|
||||
@@ -293,7 +293,6 @@ 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")
|
||||
@@ -324,4 +323,4 @@ if __name__ == "__main__":
|
||||
if failures:
|
||||
print("{} test(s) failed".format(failures))
|
||||
sys.exit(1)
|
||||
print("all tests passed")
|
||||
print("all tests passed")
|
||||
|
||||
@@ -8,12 +8,12 @@ Run:
|
||||
python3 tests/test_core_will.py
|
||||
"""
|
||||
|
||||
import copy
|
||||
import os
|
||||
import sys
|
||||
|
||||
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
|
||||
|
||||
# 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."""
|
||||
item = WillItem(_make_minimal_willitem_dict())
|
||||
# Reset STATUS to clean defaults
|
||||
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
|
||||
item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
|
||||
return item
|
||||
|
||||
|
||||
@@ -151,8 +151,8 @@ def test_will_only_valid_list():
|
||||
def _make_will_with_heirs(heirs, tx_locktime):
|
||||
"""Build a single-item will whose stored heirs == ``heirs`` and whose
|
||||
frozen tx.locktime == ``tx_locktime`` (what the will-executors hold)."""
|
||||
item = WillItem(_make_minimal_willitem_dict(heirs=copy.deepcopy(heirs)))
|
||||
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
|
||||
item = WillItem(_make_minimal_willitem_dict(heirs=copy_structure(heirs)))
|
||||
item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
|
||||
item.tx.locktime = tx_locktime
|
||||
return {"willid_1": item}
|
||||
|
||||
@@ -163,7 +163,7 @@ def test_check_heirs_unchanged_is_coherent():
|
||||
heirs = {"alice": ["addr_alice", 5000, str(lt)]}
|
||||
will = _make_will_with_heirs(heirs, lt)
|
||||
result = Will.check_willexecutors_and_heirs(
|
||||
will, copy.deepcopy(heirs), {}, False, 0, 100
|
||||
will, copy_structure(heirs), {}, False, 0, 100
|
||||
)
|
||||
assert result is True
|
||||
|
||||
|
||||
@@ -17,7 +17,6 @@ Run:
|
||||
python3 -m pytest tests/test_core_will_invalidate.py -q
|
||||
"""
|
||||
|
||||
import copy
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import MagicMock, patch
|
||||
@@ -75,7 +74,7 @@ def _make_willitem(value_sats=1000000, valid=True, extra_heirs=None):
|
||||
"change": "",
|
||||
"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.
|
||||
# Use the name-mangled attribute because tx_from_any creates a
|
||||
# Transaction whose inputs are TxInput objects; TxInput.value_sats()
|
||||
@@ -270,7 +269,7 @@ class TestInvalidateWill:
|
||||
"""
|
||||
item = _make_willitem(value_sats=100)
|
||||
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)
|
||||
|
||||
|
||||
@@ -21,7 +21,6 @@ Run:
|
||||
python3 -m pytest tests/test_group_e_karen7_invalidate.py -q
|
||||
"""
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
@@ -46,6 +45,7 @@ from electrum.transaction import (
|
||||
from electrum.util import bfh
|
||||
|
||||
from bal.core.heirs import Heirs
|
||||
from bal.core.util import copy_structure
|
||||
from bal.core.will import Will, WillItem
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
@@ -219,7 +219,7 @@ def _txs_to_will(txs, heirs_data):
|
||||
for txid, tx in txs.items():
|
||||
item_dict = {
|
||||
"tx": tx,
|
||||
"heirs": copy.deepcopy(heirs_data),
|
||||
"heirs": copy_structure(heirs_data),
|
||||
"willexecutor": None,
|
||||
"status": "",
|
||||
"description": "",
|
||||
|
||||
@@ -22,7 +22,6 @@ Run:
|
||||
python3 -m pytest tests/test_group_e_mock_karen7.py -q
|
||||
"""
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
@@ -43,6 +42,7 @@ from bal.core.reminders import (
|
||||
ical_escape,
|
||||
write_temp_ics,
|
||||
)
|
||||
from bal.core.util import copy_structure
|
||||
from bal.core.will import HeirNotFoundException, Will, WillItem
|
||||
from bal.core.willexecutors import Willexecutors
|
||||
|
||||
@@ -136,7 +136,7 @@ def _make_willitem(**overrides):
|
||||
}
|
||||
d.update(overrides)
|
||||
item = WillItem(d)
|
||||
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
|
||||
item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
|
||||
return item
|
||||
|
||||
|
||||
@@ -343,7 +343,7 @@ def test_e2_heir_change_triggers_rebuild():
|
||||
item = WillItem(
|
||||
{
|
||||
"tx": _VALID_TX_HEX,
|
||||
"heirs": copy.deepcopy(will_heirs),
|
||||
"heirs": copy_structure(will_heirs),
|
||||
"willexecutor": None,
|
||||
"status": "",
|
||||
"description": "",
|
||||
@@ -352,7 +352,7 @@ def test_e2_heir_change_triggers_rebuild():
|
||||
"baltx_fees": 100,
|
||||
}
|
||||
)
|
||||
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
|
||||
item.STATUS = WillItem.copy_status_table(WillItem.STATUS_DEFAULT)
|
||||
item.tx.locktime = lt
|
||||
will = {"willid_1": item}
|
||||
|
||||
|
||||
456
tests/test_gui_export_dialogs.py
Normal file
456
tests/test_gui_export_dialogs.py
Normal 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")
|
||||
@@ -1,19 +1,21 @@
|
||||
"""
|
||||
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.
|
||||
Covers the unified ``WillExportDialog`` (transport radios, stacked pages,
|
||||
QR build/navigation, chunk-size / autoplay, filter revert) and the unified
|
||||
``WillImportDialog`` (QR 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 json
|
||||
import sys
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
sys.path.insert(0, __file__.rsplit("/", 2)[0])
|
||||
|
||||
@@ -21,7 +23,7 @@ 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.qrtransfer import CHUNK_PRESETS, encode_transfer, split_frames
|
||||
from bal.core.will import WillItem
|
||||
|
||||
_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:
|
||||
# Smallest QR preset: long transfers produce several frames.
|
||||
QR_CHUNK_SIZE = _Cfg(CHUNK_PRESETS[0][1])
|
||||
|
||||
def read_file(self, path):
|
||||
return _PNG_BYTES
|
||||
|
||||
@@ -90,6 +103,9 @@ class StubWillItem:
|
||||
def get_status(self, name):
|
||||
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):
|
||||
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():
|
||||
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
|
||||
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
|
||||
page = d.qr_page
|
||||
assert d.transport_qr.isChecked()
|
||||
assert page.tx_strings
|
||||
assert page.frames
|
||||
assert len(page.frames) >= 1
|
||||
# Frame 1 is shown.
|
||||
assert d.qr_view.text == d.frames[0]
|
||||
assert "1" in d.progress_label.text()
|
||||
assert page.qr_view.text == page.frames[0]
|
||||
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()
|
||||
|
||||
|
||||
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
|
||||
dialogs.MessageBoxMixin.show_message = lambda self, msg, icon=None: None
|
||||
try:
|
||||
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 hasattr(d, "qr_page")
|
||||
d.close()
|
||||
finally:
|
||||
dialogs.MessageBoxMixin.show_message = orig
|
||||
@@ -144,50 +191,52 @@ def test_imported_item_status_not_none():
|
||||
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 = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
|
||||
page = d.qr_page
|
||||
assert page.fps_spin is not None
|
||||
assert not page.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")
|
||||
page.fps_spin.setValue(2)
|
||||
page._toggle_auto()
|
||||
assert page.auto_timer.isActive()
|
||||
assert page.auto_btn.text() == dialogs._("Stop")
|
||||
page._auto_step()
|
||||
assert page.index == 1
|
||||
page._toggle_auto()
|
||||
assert not page.auto_timer.isActive()
|
||||
assert page.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()
|
||||
page._toggle_auto()
|
||||
page.index = len(page.frames) - 1
|
||||
page._auto_step()
|
||||
assert not page.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
|
||||
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
|
||||
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.
|
||||
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()
|
||||
page.loop_check.setChecked(True)
|
||||
page._toggle_auto()
|
||||
assert page.auto_timer.isActive()
|
||||
page.index = len(page.frames) - 1
|
||||
page._auto_step()
|
||||
assert page.index == 0
|
||||
assert page.auto_timer.isActive()
|
||||
page._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()
|
||||
page.loop_check.setChecked(False)
|
||||
page._toggle_auto()
|
||||
page.index = len(page.frames) - 1
|
||||
page._auto_step()
|
||||
assert not page.auto_timer.isActive()
|
||||
d.close()
|
||||
|
||||
|
||||
@@ -197,18 +246,27 @@ def test_export_filter_valid_and_valid_nc():
|
||||
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
|
||||
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
|
||||
page = d.qr_page
|
||||
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).
|
||||
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).
|
||||
d._on_filter_change(2)
|
||||
assert sorted(d.tx_strings) == ["B" * 120]
|
||||
assert d.qr_view.text == d.frames[0]
|
||||
assert sorted(page.tx_strings) == ["B" * 120]
|
||||
assert page.qr_view.text == page.frames[0]
|
||||
d.close()
|
||||
|
||||
|
||||
@@ -218,116 +276,464 @@ def test_export_filter_empty_reverts():
|
||||
bw.willitems = {
|
||||
"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 = []
|
||||
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
|
||||
assert messages
|
||||
assert d._filter_index == 0 # reverted to "All"
|
||||
assert d.filter_combo.currentIndex() == 0
|
||||
assert len(d.tx_strings) == 1
|
||||
assert len(d.qr_page.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)
|
||||
d = dialogs.WillExportDialog(bw, bal_plugin=bw.bal_plugin, initial_mode="qr")
|
||||
page = d.qr_page
|
||||
first_count = len(page.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]
|
||||
assert not page.prev_btn.isEnabled()
|
||||
page._next()
|
||||
assert page.index == 1
|
||||
assert page.qr_view.text == page.frames[1]
|
||||
assert page.prev_btn.isEnabled()
|
||||
page._prev()
|
||||
assert page.index == 0
|
||||
assert page.qr_view.text == page.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
|
||||
page._on_chunk_change(len(dialogs.CHUNK_PRESETS) - 1)
|
||||
assert len(page.frames) < first_count
|
||||
assert page.index == 0
|
||||
d.close()
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# WillQrImportDialog
|
||||
# WillImportDialog (QR transport via d.qr_page)
|
||||
# ------------------------------------------------------------------ #
|
||||
|
||||
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()
|
||||
d = dialogs.WillImportDialog(bw, bal_plugin=bw.bal_plugin)
|
||||
page = d.qr_page
|
||||
assert not page.review_btn.isEnabled()
|
||||
assert not page.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()
|
||||
page._add_frame(frame)
|
||||
assert page.total == len(frames)
|
||||
assert page.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()
|
||||
assert not page.slot_area.isHidden()
|
||||
assert len(page.slot_widgets) == page.total
|
||||
assert "All" in page.status_label.text()
|
||||
# Duplicate capture is harmless.
|
||||
d._add_frame(frames[0])
|
||||
assert len(d.frames) == d.total
|
||||
page._add_frame(frames[0])
|
||||
assert len(page.frames) == page.total
|
||||
d.close()
|
||||
|
||||
|
||||
def test_import_assembles_and_decodes():
|
||||
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 = {}
|
||||
|
||||
def fake_finish(tx_strings):
|
||||
captured["tx_strings"] = tx_strings
|
||||
def fake_finish(payload):
|
||||
captured["payload"] = payload
|
||||
|
||||
d._finish_import = fake_finish
|
||||
page._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]
|
||||
page._add_frame(frame)
|
||||
page._review_and_sign()
|
||||
# The payload is rejoined into an opaque string for autodetect.
|
||||
assert captured["payload"].split("\n") == ["P" * 130, "Q" * 130]
|
||||
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():
|
||||
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 = []
|
||||
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_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)
|
||||
page._add_frame(frame)
|
||||
assert page.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])
|
||||
page._add_frame(frames_b[0])
|
||||
assert warnings
|
||||
assert d.total == 0
|
||||
assert not d.frames
|
||||
assert not d.review_btn.isEnabled()
|
||||
assert page.total == 0
|
||||
assert not page.frames
|
||||
assert not page.review_btn.isEnabled()
|
||||
d.close()
|
||||
|
||||
|
||||
def test_import_manual_entry():
|
||||
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)
|
||||
d.manual_edit.setText(frames[0])
|
||||
d._add_from_manual()
|
||||
assert d.manual_edit.text() == ""
|
||||
assert d.total == 1
|
||||
assert d.review_btn.isEnabled()
|
||||
page.manual_edit.setText(frames[0])
|
||||
page._add_from_manual()
|
||||
assert page.manual_edit.text() == ""
|
||||
assert page.total == 1
|
||||
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()
|
||||
|
||||
|
||||
|
||||
@@ -28,7 +28,6 @@ Run:
|
||||
python3 tests/test_heir_relative_anchor.py
|
||||
"""
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
@@ -40,6 +39,7 @@ from electrum import constants # noqa: E402 (path insert above)
|
||||
constants.net = constants.BitcoinRegtest
|
||||
|
||||
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
|
||||
HeirNotFoundException,
|
||||
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)."""
|
||||
d = {
|
||||
"tx": _VALID_TX_HEX,
|
||||
"heirs": copy.deepcopy(heirs),
|
||||
"heirs": copy_structure(heirs),
|
||||
"willexecutor": None,
|
||||
"status": "",
|
||||
"description": "",
|
||||
@@ -80,7 +80,7 @@ def _make_will_item(heirs, tx_locktime, status_complete=False):
|
||||
"baltx_fees": 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
|
||||
if status_complete:
|
||||
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."""
|
||||
heirs = {"alice": ["addr_alice", 5000, "1y"]}
|
||||
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
|
||||
|
||||
@@ -119,7 +119,7 @@ def test_unchanged_relative_recipe_signed_is_coherent():
|
||||
def test_unchanged_relative_recipe_unsigned_is_coherent():
|
||||
heirs = {"alice": ["addr_alice", 5000, "1y"]}
|
||||
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
|
||||
|
||||
@@ -145,7 +145,7 @@ def test_relative_recipe_shortened_on_signed_is_rebuild():
|
||||
def test_unchanged_absolute_recipe_is_coherent():
|
||||
built = {"alice": ["addr_alice", 5000, str(_FROZEN)]}
|
||||
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
|
||||
|
||||
@@ -176,6 +176,7 @@ def test_karen7_frozen_delivery_not_expired():
|
||||
valid_wid = "28b64bfd83878d15c668473aa695a2b9bc23196bc61ab3149e8f33241826978d"
|
||||
wi = WillItem(data["will"][valid_wid], _id=valid_wid)
|
||||
built_locktime = Will.get_min_locktime({valid_wid: wi})
|
||||
assert built_locktime is not None
|
||||
assert built_locktime == int(wi.tx.locktime)
|
||||
|
||||
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."""
|
||||
data = _load_karen7()
|
||||
valid_wid = "28b64bfd83878d15c668473aa695a2b9bc23196bc61ab3149e8f33241826978d"
|
||||
wi = WillItem(data["will"][valid_wid], _id=valid_wid)
|
||||
# Use _FROZEN (a UTC-midnight value) so the check is compatible with
|
||||
# the UTC anchoring code.
|
||||
frozen_locktime = _FROZEN
|
||||
|
||||
@@ -127,6 +127,11 @@ def test_sign_transactions_external_only():
|
||||
wallet=FakeWallet(),
|
||||
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)
|
||||
|
||||
|
||||
@@ -27,7 +27,6 @@ Run::
|
||||
python3 -m pytest tests/test_no_willexecutor_karen7.py -v -s
|
||||
"""
|
||||
|
||||
import copy
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
@@ -49,6 +48,7 @@ from electrum.transaction import PartialTxInput, TxOutpoint
|
||||
from electrum.util import bfh
|
||||
|
||||
from bal.core.heirs import Heirs
|
||||
from bal.core.util import copy_structure
|
||||
from bal.core.will import (
|
||||
NotCompleteWillException,
|
||||
NoWillExecutorNotPresent,
|
||||
@@ -278,11 +278,11 @@ class FakeBalWindow:
|
||||
tx["my_locktime"] = txs[txid].my_locktime
|
||||
tx["heirsvalue"] = txs[txid].heirsvalue
|
||||
tx["description"] = txs[txid].description
|
||||
tx["willexecutor"] = copy.deepcopy(txs[txid].willexecutor)
|
||||
tx["willexecutor"] = copy_structure(txs[txid].willexecutor)
|
||||
tx["status"] = "New"
|
||||
tx["baltx_fees"] = txs[txid].tx_fees
|
||||
tx["time"] = creation_time
|
||||
tx["heirs"] = copy.deepcopy(txs[txid].heirs)
|
||||
tx["heirs"] = copy_structure(txs[txid].heirs)
|
||||
tx["txchildren"] = []
|
||||
will[txid] = WillItem(tx, _id=txid, wallet=self.wallet)
|
||||
self.update_will(will)
|
||||
|
||||
@@ -7,7 +7,6 @@ but without requiring a full Qt event loop.
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
@@ -24,6 +23,7 @@ if os.path.isdir(ELECTRUM_DIR):
|
||||
|
||||
from bal.core.heirs import Heirs
|
||||
from bal.core.plugin_base import BalPlugin, BalTimestamp
|
||||
from bal.core.util import copy_structure
|
||||
from bal.core.will import (
|
||||
NoHeirsException,
|
||||
NotCompleteWillException,
|
||||
@@ -145,11 +145,11 @@ class FakeBalWindow:
|
||||
tx["my_locktime"] = txs[txid].my_locktime
|
||||
tx["heirsvalue"] = txs[txid].heirsvalue
|
||||
tx["description"] = txs[txid].description
|
||||
tx["willexecutor"] = copy.deepcopy(txs[txid].willexecutor)
|
||||
tx["willexecutor"] = copy_structure(txs[txid].willexecutor)
|
||||
tx["status"] = "New"
|
||||
tx["baltx_fees"] = txs[txid].tx_fees
|
||||
tx["time"] = creation_time
|
||||
tx["heirs"] = copy.deepcopy(txs[txid].heirs)
|
||||
tx["heirs"] = copy_structure(txs[txid].heirs)
|
||||
tx["txchildren"] = []
|
||||
will[txid] = WillItem(tx, _id=txid, wallet=self.wallet)
|
||||
Will.update_will(self.willitems, will)
|
||||
|
||||
Reference in New Issue
Block a user