kaibot/bal-electrum-plugin
1
0
Fork 0
forked from bitcoinafterlife/bal-electrum-plugin
Code Pull Requests Releases 1 Activity
Files
f4db27ca48b183bce54d26f6ed0e6ee5eb419179
bal-electrum-plugin/tests/test_inheritance_flow.py
bot 525dde2b3c add tests
2026-06-20 09:49:39 -04:00

611 lines
23 KiB
Python
Raw Blame History

"""
End-to-end inheritance flow tests (mock, no network, no daemon).
These tests exercise the real inheritance-building and state-assignment logic
of the plugin using *plausible data*:
* heir / will-executor / change addresses are genuine regtest addresses
sourced read-only from the giovanna7 wallet (see ``bal_fixtures``);
* UTXOs and txids are synthesised so the suite needs no running Electrum.
Covered scenarios (mirroring the project spec):
1. Build an inheritance with 3 heirs + 2 will-executors and verify every
heir's output value equals its resolved amount.
2. Change a heir's ADDRESS, rebuild: the new will is anticipated by 1 day,
is flagged ANTICIPATED (staying VALID), and the heir's output address is
updated.
3. Change a heir's AMOUNT, rebuild: the new will is anticipated by a further
day, flagged ANTICIPATED, and the heir's output value is updated.
4. Change a will-executor's base_fee, rebuild: the new will keeps the SAME
locktime and updates the will-executor output, while the OLD transaction
becomes UPDATED yet stays VALID.
5. Every transaction state is assigned exactly per its description.
The network is switched to regtest at import time because the giovanna7
addresses are bech32 regtest addresses; this must happen before importing the
bal modules (``BalPlugin.chainname`` is computed at import).
"""
import itertools
import os
import sys
import time
import copy
import pytest
from unittest.mock import MagicMock, patch
from electrum import constants
sys.path.insert(0, os.path.dirname(__file__))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
@pytest.fixture(autouse=True)
def _regtest_network():
"""Switch the global Electrum network to regtest for these tests.
The giovanna7 addresses are bech32 *regtest* addresses, so the network must
be regtest while building/validating transactions. The previous network is
restored afterwards so this module never pollutes mainnet-based tests that
may run later in the same pytest process.
"""
previous = constants.net
constants.BitcoinRegtest.set_as_network()
try:
yield
finally:
previous.set_as_network()
import bal_fixtures as fx # noqa: E402
from bal.core.heirs import ( # noqa: E402
Heirs,
HEIR_ADDRESS,
HEIR_AMOUNT,
HEIR_LOCKTIME,
HEIR_REAL_AMOUNT,
)
from bal.core.will import Will, WillItem # noqa: E402
from bal.core.util import Util # noqa: E402
from bal.core.willexecutors import Willexecutors # noqa: E402
from electrum.transaction import PartialTransaction # noqa: E402
DAY = 86400
# ------------------------------------------------------------------ #
# Helpers
# ------------------------------------------------------------------ #
def _make_heirs(entries):
"""Build a ``Heirs`` instance (bypassing the wallet-backed ``__init__``).
``entries`` is ``{name: [address, amount, locktime]}``.
"""
h = Heirs.__new__(Heirs)
dict.__init__(h)
h.db = MagicMock()
h.wallet = MagicMock()
h.update(copy.deepcopy(entries))
return h
def _build(heirs, willexecutors, *, utxo_value=5_000_000, utxo_count=4,
from_locktime=None, tx_fees=100, no_willexecutor=False):
"""Run the real builder and return the produced ``{txid: tx}`` mapping.
``PartialTransaction.txid`` is patched with a deterministic counter because
the UTXOs are unsigned synthetic inputs whose real txid would be ``None``
(this is the same technique used by the other mock tests).
"""
wallet = fx.fake_wallet()
utxos = fx.make_utxos(utxo_count, utxo_value)
plugin = fx.fake_bal_plugin(willexecutors, no_willexecutor=no_willexecutor)
if from_locktime is None:
from_locktime = int(time.time())
counter = itertools.count(1)
def fake_txid(self):
# Stable per-object txid: cache on the instance so repeated calls
# within the builder return the same value, but distinct tx objects
# get distinct ids.
if not hasattr(self, "_test_txid"):
self._test_txid = f"{next(counter):064x}"
return self._test_txid
with patch.object(Willexecutors, "get_willexecutors",
return_value=willexecutors), \
patch.object(PartialTransaction, "txid", fake_txid):
txs = heirs.get_transactions(plugin, wallet, tx_fees, utxos,
from_locktime)
return txs or {}
def _willexecutors(specs):
"""Build a ``{url: we_dict}`` mapping from ``[(url, address, base_fee)]``."""
out = {}
for url, address, base_fee in specs:
out[url] = {
"url": url,
"address": address,
"base_fee": base_fee,
"selected": True,
"status": 200,
}
return out
def _heir_output(tx, address):
"""Return the output value paid to ``address`` in ``tx`` (or None)."""
for o in tx.outputs():
if o.address == address:
return o.value
return None
def _backup_tx(txs):
"""Return the backup transaction (the one without a will-executor)."""
for tx in txs.values():
if not getattr(tx, "willexecutor", None):
return tx
return None
def _tx_for_we(txs, url):
"""Return the transaction built for the will-executor ``url``."""
for tx in txs.values():
we = getattr(tx, "willexecutor", None)
if we and we.get("url") == url:
return tx
return None
# ------------------------------------------------------------------ #
# Test 1: amounts match heir outputs
# ------------------------------------------------------------------ #
def test_inheritance_amounts_match_outputs():
"""Each heir's resolved amount equals its output value in every built tx."""
now = int(time.time())
lt = now + 365 * DAY
ha = fx.heir_addresses(3)
wea = fx.willexecutor_addresses(2)
heirs = _make_heirs({
"alice": [ha[0], "20%", str(lt)],
"bob": [ha[1], "30%", str(lt)],
"carol": [ha[2], "50%", str(lt)],
})
wes = _willexecutors([
("https://we1.example", wea[0], 100000),
("https://we2.example", wea[1], 50000),
])
txs = _build(heirs, wes, no_willexecutor=True)
assert txs, "no transactions built"
# We expect one tx per selected will-executor plus one backup tx.
assert _tx_for_we(txs, "https://we1.example") is not None
assert _tx_for_we(txs, "https://we2.example") is not None
assert _backup_tx(txs) is not None
for txid, tx in txs.items():
for name, hd in getattr(tx, "heirs", {}).items():
real = hd[HEIR_REAL_AMOUNT]
if isinstance(real, str) and "DUST" in real:
continue
out_val = _heir_output(tx, hd[HEIR_ADDRESS])
assert out_val == real, (
f"{name}: output {out_val} != resolved amount {real} "
f"in tx {txid[:8]}"
)
# The will-executor fee output, if any, must equal its base_fee.
we = getattr(tx, "willexecutor", None)
if we:
assert _heir_output(tx, we["address"]) == we["base_fee"], (
f"will-executor {we['url']} output != base_fee"
)
# ------------------------------------------------------------------ #
# Test 2 + 3 + 4: state transitions on update
# ------------------------------------------------------------------ #
def _single_we_will(heirs, url, address, base_fee, *, from_locktime, time_stamp,
txid=None):
"""Build a one-will-executor will and wrap it into a {txid: WillItem}.
``txid`` lets the caller pin a stable, unique transaction id (the synthetic
unsigned tx has no real txid). Returns ``(will_dict, txid, tx)`` for the
single non-backup transaction.
"""
wes = _willexecutors([(url, address, base_fee)])
txs = _build(heirs, wes, from_locktime=from_locktime, no_willexecutor=False)
tx = _tx_for_we(txs, url)
assert tx is not None, "no will-executor transaction built"
if txid is None:
txid = tx._test_txid if hasattr(tx, "_test_txid") else tx.txid()
item = WillItem.__new__(WillItem)
item.tx = tx
item._id = txid
item.heirs = copy.deepcopy(tx.heirs)
item.we = copy.deepcopy(tx.willexecutor)
item.status = ""
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
item.tx_fees = 100
item.time = time_stamp
item.father = None
item.children = {}
item.description = ""
item.change = ""
item.set_status("VALID", True)
return {txid: item}, txid, tx
def test_change_heir_address_anticipates_one_day():
"""Changing a heir's address rebuilds an anticipated (1 day) ANTICIPATED tx."""
now = int(time.time())
lt = int((now + 365 * DAY) / DAY) * DAY # midnight-aligned timestamp
ha = fx.heir_addresses(4)
# Original will.
heirs = _make_heirs({"alice": [ha[0], "100%", str(lt)]})
old_will, old_txid, old_tx = _single_we_will(
heirs, "https://we.example", ha[3], 100000,
from_locktime=now, time_stamp=now - 10,
)
old_item = old_will[old_txid]
old_locktime = old_item.tx.locktime
assert _heir_output(old_tx, ha[0]) is not None
# Rebuild with the heir's address changed.
new_heirs = _make_heirs({"alice": [ha[1], "100%", str(lt)]})
new_will, new_txid, new_tx = _single_we_will(
new_heirs, "https://we.example", ha[3], 100000,
from_locktime=now, time_stamp=now,
)
new_item = new_will[new_txid]
# The new will item, compared against the old one sharing the same heirs
# set but a different destination, should anticipate by 1 day.
anticipated = new_item.set_anticipate(old_item)
expected = int(Util.anticipate_locktime(old_locktime, days=1))
assert anticipated, "new will should be anticipated"
assert new_item.get_status("ANTICIPATED"), "ANTICIPATED must be set"
assert new_item.get_status("VALID"), "VALID must remain set"
assert int(new_item.tx.locktime) == expected, (
f"locktime {new_item.tx.locktime} != anticipated {expected}"
)
# The new transaction pays the NEW address, not the old one.
assert _heir_output(new_tx, ha[1]) is not None
assert _heir_output(new_tx, ha[0]) is None
def test_change_heir_amount_anticipates_one_day():
"""Changing a heir's amount rebuilds an anticipated (1 day) ANTICIPATED tx."""
now = int(time.time())
lt = int((now + 365 * DAY) / DAY) * DAY
ha = fx.heir_addresses(4)
heirs = _make_heirs({
"alice": [ha[0], "60%", str(lt)],
"bob": [ha[1], "40%", str(lt)],
})
old_will, old_txid, old_tx = _single_we_will(
heirs, "https://we.example", ha[3], 100000,
from_locktime=now, time_stamp=now - 10,
)
old_item = old_will[old_txid]
old_locktime = old_item.tx.locktime
old_alice_out = _heir_output(old_tx, ha[0])
# Rebuild with alice's amount changed.
new_heirs = _make_heirs({
"alice": [ha[0], "80%", str(lt)],
"bob": [ha[1], "20%", str(lt)],
})
new_will, new_txid, new_tx = _single_we_will(
new_heirs, "https://we.example", ha[3], 100000,
from_locktime=now, time_stamp=now,
)
new_item = new_will[new_txid]
anticipated = new_item.set_anticipate(old_item)
expected = int(Util.anticipate_locktime(old_locktime, days=1))
assert anticipated, "new will should be anticipated"
assert new_item.get_status("ANTICIPATED"), "ANTICIPATED must be set"
assert new_item.get_status("VALID"), "VALID must remain set"
assert int(new_item.tx.locktime) == expected
# alice's output value changed (80% > 60%).
new_alice_out = _heir_output(new_tx, ha[0])
assert new_alice_out is not None and new_alice_out != old_alice_out, (
f"alice output should change: old {old_alice_out} new {new_alice_out}"
)
def test_change_we_base_fee_keeps_locktime_and_marks_updated():
"""Changing only the will-executor base_fee keeps the locktime; old tx UPDATED.
Per the spec: same locktime + same heirs but a new transaction -> the old
transaction is flagged UPDATED while keeping VALID; the new transaction's
will-executor output reflects the new base_fee.
"""
now = int(time.time())
lt = int((now + 365 * DAY) / DAY) * DAY
ha = fx.heir_addresses(4)
we_addr = ha[3]
heirs = _make_heirs({"alice": [ha[0], "100%", str(lt)]})
old_will, old_txid, old_tx = _single_we_will(
heirs, "https://we.example", we_addr, 100000,
from_locktime=now, time_stamp=now - 10, txid="d1" * 32,
)
old_item = old_will[old_txid]
old_locktime = int(old_item.tx.locktime)
# Rebuild with a HIGHER base_fee only.
new_will, new_txid, new_tx = _single_we_will(
heirs, "https://we.example", we_addr, 200000,
from_locktime=now, time_stamp=now, txid="d2" * 32,
)
new_item = new_will[new_txid]
# check_anticipate must keep the same locktime for a base_fee increase.
kept = Will.check_anticipate(old_item, new_item)
assert int(kept) == old_locktime, (
f"locktime should stay {old_locktime}, got {kept}"
)
# set_anticipate returns False (no anticipation) since locktime is unchanged.
assert not new_item.set_anticipate(old_item), (
"no anticipation expected for a base_fee-only change"
)
assert int(new_item.tx.locktime) == old_locktime
# The new will-executor output reflects the new base_fee.
assert _heir_output(new_tx, we_addr) == 200000
# Now place both transactions in one will (they share the same wallet UTXO)
# and let the state engine flag the old one as UPDATED while keeping VALID.
will = {old_txid: old_item, new_txid: new_item}
all_inputs = Will.get_all_inputs(will, only_valid=True)
Will.search_updated(all_inputs)
assert old_item.get_status("UPDATED"), "old tx must be UPDATED"
assert old_item.get_status("VALID"), "old tx must remain VALID"
assert not new_item.get_status("UPDATED"), "new tx must NOT be UPDATED"
assert new_item.get_status("VALID"), "new tx must remain VALID"
# ------------------------------------------------------------------ #
# Test 5: each state assigned per its description
# ------------------------------------------------------------------ #
def _wi(tx_locktime, heirs_data, txid, *, time_stamp=0):
"""Build a minimal VALID WillItem around a real PartialTransaction.
The funding UTXO uses a txid *different* from this will item's own ``txid``
so ``add_willtree`` does not mistake the item for its own parent.
"""
utxo = fx.make_utxo("9e" * 32, 200_000)
from electrum.transaction import PartialTxOutput
addr = heirs_data[next(iter(heirs_data))][HEIR_ADDRESS]
tx = PartialTransaction.from_io(
[utxo],
[PartialTxOutput.from_address_and_value(addr, 100_000)],
locktime=tx_locktime, version=2,
)
item = WillItem.__new__(WillItem)
item.tx = tx
item._id = txid
item.heirs = heirs_data
item.we = None
item.status = ""
item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
item.tx_fees = 100
item.time = time_stamp
item.father = None
item.children = {}
item.description = ""
item.change = ""
item.set_status("VALID", True)
return item
def test_state_anticipated_keeps_valid():
now = int(time.time())
lt = int((now + 30 * DAY) / DAY) * DAY
ha = fx.heir_addresses(2)
old = _wi(lt, {"a": [ha[0], 100_000, str(lt)]}, "aa" * 32, time_stamp=now - 5)
new = _wi(lt, {"a": [ha[1], 100_000, str(lt)]}, "bb" * 32, time_stamp=now)
assert new.set_anticipate(old)
assert new.get_status("ANTICIPATED")
assert new.get_status("VALID") # VALID is NOT cleared
def test_state_replaced_clears_valid_and_propagates():
"""REPLACED: input spent by a lower-locktime tx; cascades to children."""
now = int(time.time())
lt_parent = now + 60 * DAY
lt_child = now + 30 * DAY
lt_replacer = now + 15 * DAY
ha = fx.heir_addresses(3)
from electrum.transaction import PartialTxOutput, PartialTxInput, TxOutpoint
from electrum.util import bfh
parent_txid = "aa" * 32
parent_tx = PartialTransaction.from_io(
[fx.make_utxo("aa" * 32, 200_000)],
[PartialTxOutput.from_address_and_value(ha[0], 100_000),
PartialTxOutput.from_address_and_value(fx.change_address(), 99_000)],
locktime=lt_parent, version=2,
)
pwi = WillItem.__new__(WillItem)
pwi.tx = parent_tx; pwi._id = parent_txid
pwi.heirs = {"a": [ha[0], 100_000, str(lt_parent)]}; pwi.we = None
pwi.status = ""; pwi.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
pwi.tx_fees = 100; pwi.time = now; pwi.father = None; pwi.children = {}
pwi.description = ""; pwi.change = ""
pwi.set_status("VALID", True)
ci = PartialTxInput(prevout=TxOutpoint(txid=bfh(parent_txid), out_idx=1))
ci._trusted_value_sats = 99_000; ci.is_mine = True
child_tx = PartialTransaction.from_io(
[ci], [PartialTxOutput.from_address_and_value(ha[1], 50_000)],
locktime=lt_child, version=2,
)
cwi = WillItem.__new__(WillItem)
cwi.tx = child_tx; cwi._id = "bb" * 32
cwi.heirs = {"b": [ha[1], 50_000, str(lt_child)]}; cwi.we = None
cwi.status = ""; cwi.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
cwi.tx_fees = 100; cwi.time = now; cwi.father = None; cwi.children = {}
cwi.description = ""; cwi.change = ""
cwi.set_status("VALID", True)
replacer_tx = PartialTransaction.from_io(
[fx.make_utxo("aa" * 32, 200_000)],
[PartialTxOutput.from_address_and_value(ha[2], 150_000)],
locktime=lt_replacer, version=2,
)
rwi = WillItem.__new__(WillItem)
rwi.tx = replacer_tx; rwi._id = "cc" * 32
rwi.heirs = {"c": [ha[2], 150_000, str(lt_replacer)]}; rwi.we = None
rwi.status = ""; rwi.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
rwi.tx_fees = 100; rwi.time = now; rwi.father = None; rwi.children = {}
rwi.description = ""; rwi.change = ""
rwi.set_status("VALID", True)
will = {pwi._id: pwi, cwi._id: cwi, rwi._id: rwi}
Will.add_willtree(will)
Will.search_rai(Will.get_all_inputs(will, only_valid=True),
[fx.make_utxo("aa" * 32, 200_000)], will, fx.fake_wallet())
assert pwi.get_status("REPLACED") and not pwi.get_status("VALID")
assert cwi.get_status("REPLACED") and not cwi.get_status("VALID")
assert rwi.get_status("VALID") and not rwi.get_status("REPLACED")
def test_state_invalidated_clears_valid():
"""INVALIDATED: input spent elsewhere and parent tx not in the will."""
now = int(time.time())
lt = now + 30 * DAY
ha = fx.heir_addresses(1)
from electrum.transaction import PartialTxOutput, PartialTxInput, TxOutpoint
from electrum.util import bfh
txin = PartialTxInput(prevout=TxOutpoint(txid=bfh("dd" * 32), out_idx=0))
txin._trusted_value_sats = 100_000; txin.is_mine = True
tx = PartialTransaction.from_io(
[txin], [PartialTxOutput.from_address_and_value(ha[0], 90_000)],
locktime=lt, version=2,
)
wi = WillItem.__new__(WillItem)
wi.tx = tx; wi._id = "ee" * 32
wi.heirs = {"a": [ha[0], 90_000, str(lt)]}; wi.we = None
wi.status = ""; wi.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
wi.tx_fees = 100; wi.time = now; wi.father = None; wi.children = {}
wi.description = ""; wi.change = ""
wi.set_status("VALID", True)
will = {wi._id: wi}
Will.add_willtree(will)
# all_utxos empty -> input is gone; wallet has no record of our tx.
Will.search_rai(Will.get_all_inputs(will, only_valid=True), [], will,
fx.fake_wallet())
assert wi.get_status("INVALIDATED")
assert not wi.get_status("VALID")
def test_state_pending_clears_valid():
"""PENDING: our tx is seen unconfirmed in the wallet (height 0)."""
now = int(time.time())
lt = now + 30 * DAY
ha = fx.heir_addresses(1)
wi = _wi(lt, {"a": [ha[0], 100_000, str(lt)]}, "ab" * 32, time_stamp=now)
wallet = fx.fake_wallet()
wallet.get_tx_info.return_value.tx_mined_status.height.return_value = 0
will = {wi._id: wi}
Will.add_willtree(will)
# utxos_list empty -> the input is spent (our tx is in mempool spending it).
Will.check_invalidated(will, [], wallet)
assert wi.get_status("PENDING")
assert not wi.get_status("VALID")
def test_state_confirmed_clears_valid():
"""CONFIRMED: our tx is mined (height > 0)."""
now = int(time.time())
lt = now + 30 * DAY
ha = fx.heir_addresses(1)
wi = _wi(lt, {"a": [ha[0], 100_000, str(lt)]}, "ac" * 32, time_stamp=now)
wallet = fx.fake_wallet()
wallet.get_tx_info.return_value.tx_mined_status.height.return_value = 123
will = {wi._id: wi}
Will.add_willtree(will)
Will.check_invalidated(will, [], wallet)
assert wi.get_status("CONFIRMED")
assert not wi.get_status("VALID")
def test_state_updated_keeps_valid():
"""UPDATED: same locktime + same heirs, older tx flagged UPDATED, stays VALID."""
now = int(time.time())
lt = int((now + 365 * DAY) / DAY) * DAY
ha = fx.heir_addresses(1)
from electrum.transaction import PartialTxOutput
shared_utxo = lambda: fx.make_utxo("fa" * 32, 500_000)
older = WillItem.__new__(WillItem)
older.tx = PartialTransaction.from_io(
[shared_utxo()],
[PartialTxOutput.from_address_and_value(ha[0], 100_000),
PartialTxOutput.from_address_and_value(fx.willexecutor_addresses(1)[0], 100_000)],
locktime=lt, version=2)
older._id = "f1" * 32
older.heirs = {"a": [ha[0], 100_000, str(lt)]}
older.we = {"url": "https://we.example", "base_fee": 100000}
older.status = ""; older.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
older.tx_fees = 100; older.time = now - 100; older.father = None
older.children = {}; older.description = ""; older.change = ""
older.set_status("VALID", True)
newer = WillItem.__new__(WillItem)
newer.tx = PartialTransaction.from_io(
[shared_utxo()],
[PartialTxOutput.from_address_and_value(ha[0], 100_000),
PartialTxOutput.from_address_and_value(fx.willexecutor_addresses(1)[0], 200_000)],
locktime=lt, version=2)
newer._id = "f2" * 32
newer.heirs = {"a": [ha[0], 100_000, str(lt)]}
newer.we = {"url": "https://we.example", "base_fee": 200000}
newer.status = ""; newer.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
newer.tx_fees = 100; newer.time = now; newer.father = None
newer.children = {}; newer.description = ""; newer.change = ""
newer.set_status("VALID", True)
will = {older._id: older, newer._id: newer}
Will.search_updated(Will.get_all_inputs(will, only_valid=True))
assert older.get_status("UPDATED") and older.get_status("VALID")
assert not newer.get_status("UPDATED") and newer.get_status("VALID")
# ------------------------------------------------------------------ #
if __name__ == "__main__":
import inspect
for name, obj in sorted(globals().items()):
if name.startswith("test_") and inspect.isfunction(obj):
obj()
print(f" [OK] {name}")
print("[OK] All inheritance-flow tests passed")
View Git Blame Copy Permalink