Add Willexecutors.is_valid + grey italic styling for invalid executors

2026-07-30 17:49:28 -04:00
parent 4a9299d85b
commit 9bf088b7ff
20 changed files with 3826 additions and 66 deletions
--- a/tests/test_core_will_invalidate.py
+++ b/tests/test_core_will_invalidate.py
@@ -0,0 +1,484 @@
+"""
+Tests for will invalidation (cancellation) in ``bal.core.will``.
+
+Covers:
+* Will.invalidate_will() - building the invalidation transaction
+* Will.set_invalidate() - marking will items as invalidated (status cascade)
+
+The invalidation ("cancellation") transaction spends the same UTXOs that were
+committed to the time-locked will, making the original will transactions
+unspendable.  This is the mechanism used when:
+  * The will expires (locktime in the past)
+  * The owner postpones a signed/sent will to a later date
+  * The check-alive threshold is passed (dead-man's switch)
+
+Run:
+    QT_QPA_PLATFORM=offscreen PYTHONPATH=electrum-src \
+        python3 -m pytest tests/test_core_will_invalidate.py -q
+"""
+
+import copy
+import os
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
+
+from bal.core.will import Will, WillItem
+
+# Patch Transaction.add_info_from_wallet so WillItem can parse the tx hex
+# without a live Electrum wallet connection.
+from electrum.transaction import Transaction
+_patcher = patch.object(Transaction, "add_info_from_wallet")
+_patcher.start()
+
+# A valid serialized Bitcoin transaction hex (1 input + 1 P2PKH output,
+# version 2).  Reused across multiple test suites.
+_VALID_TX_HEX = (
+    "01000000012a5c9a94fcde98f5581cd00162c60a13936ceb75389ea65b"
+    "f38633b424eb4031000000006c493046022100a82bbc57a0136751e543"
+    "3f41cf000b3f1a99c6744775e76ec764fb78c54ee100022100f9e80b7d"
+    "e89de861dc6fb0c1429d5da72c2b6b2ee2406bc9bfb1beedd729d98501"
+    "2102e61d176da16edd1d258a200ad9759ef63adf8e14cd97f53227bae3"
+    "5cdb84d2f6ffffffff0140420f00000000001976a914230ac37834073a"
+    "42146f11ef8414ae929feaafc388ac00000000"
+)
+
+# The prevout string that _VALID_TX_HEX spends (input 0).
+_PREVOUT_STR = "3140eb24b43386f35ba69e3875eb6c93130ac66201d01c58f598defc949a5c2a:0"
+
+# Change address for the invalidation output.
+_CHANGE_ADDR = "14CHYaaByjJZpx4oHBpfDMdqhTyXnZ3kVs"
+
+
+# ------------------------------------------------------------------ #
+# Helpers
+# ------------------------------------------------------------------ #
+
+def _make_willitem(value_sats=1000000, valid=True, extra_heirs=None):
+    """Create a WillItem from _VALID_TX_HEX with a known input value.
+
+    The input's ``_trusted_value_sats`` is set so that
+    ``invalidate_will`` can read the balance from it.
+    """
+    heirs = {"alice": ["addr_alice", 5000, "30d"]}
+    if extra_heirs:
+        heirs.update(extra_heirs)
+    item = WillItem({
+        "tx": _VALID_TX_HEX,
+        "heirs": heirs,
+        "willexecutor": None,
+        "status": "",
+        "description": "",
+        "time": 0,
+        "change": "",
+        "baltx_fees": 100,
+    })
+    item.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT)
+    # Set the input value so the balance calculation works.
+    item.tx.inputs()[0]._trusted_value_sats = value_sats
+    if not valid:
+        item.set_status("INVALIDATED", True)
+    return item
+
+
+def _make_utxo(prevout_str=None, value_sats=1000000, is_coinbase=False):
+    """Create a minimal mock UTXO (wallet-side) matching a will input."""
+    if prevout_str is None:
+        prevout_str = _PREVOUT_STR
+    utxo = MagicMock()
+    utxo.prevout.to_str.return_value = prevout_str
+    utxo.is_coinbase_output.return_value = is_coinbase
+    utxo.block_height = 1
+    utxo.value_sats.return_value = value_sats
+    return utxo
+
+
+def _mock_wallet(utxos, change_addr=_CHANGE_ADDR):
+    """Create a mock wallet with the given UTXOs and change address."""
+    wallet = MagicMock()
+    wallet.get_utxos.return_value = utxos
+    wallet.get_change_addresses_for_new_transaction.return_value = [change_addr]
+    wallet.network = MagicMock()
+    return wallet
+
+
+def _run_invalidate(will, wallet, fees_per_byte=10, current_height=800000):
+    """Run ``Will.invalidate_will`` with mocked Electrum tx building.
+
+    Returns ``(result, mock_from_io, mock_out)`` so tests can inspect
+    the calls to ``PartialTransaction.from_io`` and
+    ``PartialTxOutput.from_address_and_value``.
+    """
+    mock_output = MagicMock()
+    mock_output.value = 0
+    mock_output.is_change = False
+
+    mock_tx = MagicMock()
+    mock_tx.txid.return_value = "invalidation_txid"
+    mock_tx.estimated_size.return_value = 200
+
+    with patch("bal.core.will.Util.get_current_height", return_value=current_height), \
+         patch("electrum.transaction.PartialTxOutput.from_address_and_value",
+               return_value=mock_output) as mock_out, \
+         patch("electrum.transaction.PartialTransaction.from_io",
+               return_value=mock_tx) as mock_from_io:
+        result = Will.invalidate_will(will, wallet, fees_per_byte)
+        return result, mock_from_io, mock_out
+
+
+# ================================================================== #
+# Will.invalidate_will - building the cancellation transaction
+# ================================================================== #
+
+class TestInvalidateWill:
+    """Tests for ``Will.invalidate_will()``: the cancellation transaction."""
+
+    def test_basic_returns_tx(self):
+        """A single valid will item with a matching wallet UTXO produces an
+        invalidation transaction."""
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo()])
+
+        result, mock_from_io, _ = _run_invalidate(will, wallet, fees_per_byte=10)
+
+        assert result is not None, "should return a transaction"
+
+    def test_basic_rbf_enabled(self):
+        """The invalidation tx has RBF (Replace-By-Fee) enabled."""
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo()])
+
+        result, _, _ = _run_invalidate(will, wallet)
+
+        result.set_rbf.assert_called_with(True)
+
+    def test_basic_locktime_is_current_height(self):
+        """The invalidation tx locktime equals the current block height."""
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo()])
+        current_height = 750000
+
+        _, mock_from_io, _ = _run_invalidate(will, wallet, current_height=current_height)
+
+        # from_io(inputs, outputs, locktime=<height>, version=2)
+        _, kwargs = mock_from_io.call_args
+        assert kwargs["locktime"] == current_height
+
+    def test_basic_version_2(self):
+        """The invalidation tx uses Bitcoin transaction version 2."""
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo()])
+
+        _, mock_from_io, _ = _run_invalidate(will, wallet)
+
+        _, kwargs = mock_from_io.call_args
+        assert kwargs["version"] == 2
+
+    def test_basic_output_value_deducts_fee(self):
+        """The invalidation output value is balance minus fee.
+
+        Fee = estimated_size * fees_per_byte.
+        """
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo()])
+        fees_per_byte = 10
+
+        _, mock_from_io, mock_out = _run_invalidate(will, wallet, fees_per_byte=fees_per_byte)
+
+        # The second call to from_address_and_value uses balance - fee.
+        # estimated_size returns 200, so fee = 200 * 10 = 2000.
+        # Expected output value = 1000000 - 2000 = 998000.
+        second_call_value = mock_out.call_args_list[1][0][1]
+        assert second_call_value == 998000
+
+    def test_basic_spends_correct_utxos(self):
+        """The invalidation tx spends the same UTXOs as the will."""
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo()])
+
+        _, mock_from_io, _ = _run_invalidate(will, wallet)
+
+        # First positional arg is the list of UTXOs to spend.
+        spent_utxos = mock_from_io.call_args[0][0]
+        assert len(spent_utxos) == 1
+        assert spent_utxos[0].prevout.to_str() == _PREVOUT_STR
+
+    def test_no_matching_utxos_returns_none(self):
+        """When wallet UTXOs don't match any will inputs, returns None."""
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo(prevout_str="aaaa:1")])
+
+        result, _, _ = _run_invalidate(will, wallet)
+        assert result is None
+
+    def test_no_valid_items_returns_none(self):
+        """When all will items are INVALIDATED, returns None."""
+        item = _make_willitem(valid=False)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo()])
+
+        result, _, _ = _run_invalidate(will, wallet)
+        assert result is None
+
+    def test_empty_will_returns_none(self):
+        """An empty will dictionary returns None."""
+        wallet = _mock_wallet([_make_utxo()])
+
+        result, _, _ = _run_invalidate({}, wallet)
+        assert result is None
+
+    def test_skips_young_coinbase(self):
+        """Coinbase UTXOs younger than current_height + 100 are skipped."""
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        # Coinbase UTXO: block_height = 800050, current_height = 800000
+        # 800050 < 800000 + 100 => skipped
+        utxo = _make_utxo(value_sats=1000000, is_coinbase=True)
+        utxo.block_height = 800050
+        wallet = _mock_wallet([utxo])
+
+        result, _, _ = _run_invalidate(will, wallet, current_height=800000)
+        assert result is None
+
+    def test_includes_mature_coinbase(self):
+        """Coinbase UTXOs at or above current_height + 100 are included."""
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        utxo = _make_utxo(value_sats=1000000, is_coinbase=True)
+        utxo.block_height = 800150  # >= 800000 + 100
+        wallet = _mock_wallet([utxo])
+
+        result, _, _ = _run_invalidate(will, wallet, current_height=800000)
+        assert result is not None
+
+    def test_fee_exceeds_balance_returns_none(self):
+        """When the fee exceeds the balance, returns None.
+
+        estimated_size (200) * fees_per_byte (100) = 20000 > balance (100).
+        """
+        item = _make_willitem(value_sats=100)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo()])
+
+        result, mock_from_io, _ = _run_invalidate(will, wallet, fees_per_byte=100)
+
+        assert result is None
+        # from_io is still called once (for fee estimation), but the
+        # result is discarded because balance - fee <= 0.
+        assert mock_from_io.call_count == 1
+
+    def test_only_valid_items_contribute_balance(self):
+        """INVALIDATED will items are excluded from the balance."""
+        valid_item = _make_willitem(value_sats=1000000, valid=True)
+        invalid_item = _make_willitem(value_sats=2000000, valid=False)
+        will = {"valid": valid_item, "invalid": invalid_item}
+        wallet = _mock_wallet([_make_utxo()])
+
+        _, mock_from_io, mock_out = _run_invalidate(will, wallet, fees_per_byte=10)
+
+        # Balance = 1000000 (valid only), fee = 200 * 10 = 2000
+        # Output value = 998000
+        second_call_value = mock_out.call_args_list[1][0][1]
+        assert second_call_value == 998000
+
+    def test_first_from_io_uses_full_balance(self):
+        """The first from_io call uses the full balance (before fee deduction)
+        to estimate the fee."""
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo()])
+
+        _, mock_from_io, mock_out = _run_invalidate(will, wallet, fees_per_byte=10)
+
+        # First from_address_and_value call: value = balance (1000000)
+        first_call_value = mock_out.call_args_list[0][0][1]
+        assert first_call_value == 1000000
+
+    def test_output_address_is_change_address(self):
+        """The invalidation output goes to the wallet's change address."""
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo()])
+
+        _, _, mock_out = _run_invalidate(will, wallet)
+
+        # Both calls to from_address_and_value use the change address.
+        for call in mock_out.call_args_list:
+            assert call[0][0] == _CHANGE_ADDR
+
+    def test_multiple_utxos_all_matched(self):
+        """Multiple matching UTXOs are all included in the invalidation."""
+        item1 = _make_willitem(value_sats=500000)
+        item2 = _make_willitem(value_sats=300000)
+        will = {"tx1": item1, "tx2": item2}
+
+        # Two UTXOs with different prevouts matching the two will items.
+        # Since both items use the same _VALID_TX_HEX, their prevout is the
+        # same.  To test multiple UTXOs, we need a second tx hex with a
+        # different input.
+        #
+        # However, get_all_inputs deduplicates by prevout_str, so even with
+        # two items sharing the same prevout, only one entry is added to
+        # prevout_to_spend.  The first matching UTXO is what matters.
+        utxos = [_make_utxo()]
+        wallet = _mock_wallet(utxos)
+
+        result, mock_from_io, _ = _run_invalidate(will, wallet)
+        assert result is not None
+        # Only 1 UTXO spent (deduplication of shared prevout)
+        spent_utxos = mock_from_io.call_args[0][0]
+        assert len(spent_utxos) == 1
+
+    def test_zero_fees_per_byte(self):
+        """With zero fee rate, the full balance goes to the output."""
+        item = _make_willitem(value_sats=1000000)
+        will = {"willtxid1": item}
+        wallet = _mock_wallet([_make_utxo()])
+
+        _, mock_from_io, mock_out = _run_invalidate(will, wallet, fees_per_byte=0)
+
+        assert mock_from_io.call_count == 2  # two calls (both succeed)
+        # Output value = balance - 0 = 1000000
+        second_call_value = mock_out.call_args_list[1][0][1]
+        assert second_call_value == 1000000
+
+
+# ================================================================== #
+# Will.set_invalidate - status flag cascade
+# ================================================================== #
+
+class TestSetInvalidate:
+    """Tests for ``Will.set_invalidate()``: marking will items as invalidated."""
+
+    def test_single_item_no_children(self):
+        """Invalidating a single will item sets INVALIDATED and clears VALID."""
+        item = _make_willitem(valid=True)
+        item.children = {}
+        will = {"willid1": item}
+
+        Will.set_invalidate("willid1", will)
+
+        assert item.get_status("INVALIDATED") is True
+        assert item.get_status("VALID") is False
+
+    def test_cascades_to_direct_children(self):
+        """Invalidating a parent cascades INVALIDATED to its children."""
+        parent = _make_willitem(valid=True)
+        child = _make_willitem(valid=True)
+
+        parent.children = {"child_id": ["child_id", 0, 0]}
+        child.children = {}
+
+        will = {"parent_id": parent, "child_id": child}
+
+        Will.set_invalidate("parent_id", will)
+
+        assert parent.get_status("INVALIDATED") is True
+        assert parent.get_status("VALID") is False
+        assert child.get_status("INVALIDATED") is True
+        assert child.get_status("VALID") is False
+
+    def test_cascades_to_grandchildren(self):
+        """Invalidating cascades through multiple levels of descendants."""
+        root = _make_willitem(valid=True)
+        branch = _make_willitem(valid=True)
+        leaf = _make_willitem(valid=True)
+
+        root.children = {"branch_id": ["branch_id", 0, 0]}
+        branch.children = {"leaf_id": ["leaf_id", 0, 0]}
+        leaf.children = {}
+
+        will = {
+            "root_id": root,
+            "branch_id": branch,
+            "leaf_id": leaf,
+        }
+
+        Will.set_invalidate("root_id", will)
+
+        for name, item in [("root", root), ("branch", branch), ("leaf", leaf)]:
+            assert item.get_status("INVALIDATED") is True, f"{name} should be INVALIDATED"
+            assert item.get_status("VALID") is False, f"{name} should not be VALID"
+
+    def test_empty_children_dict(self):
+        """A will item with an empty children dict is a leaf (no cascade)."""
+        item = _make_willitem(valid=True)
+        item.children = {}
+        will = {"wid": item}
+
+        Will.set_invalidate("wid", will)
+
+        assert item.get_status("INVALIDATED") is True
+        assert item.get_status("VALID") is False
+
+    def test_does_not_affect_siblings(self):
+        """Invalidating one item does not affect unrelated siblings."""
+        item_a = _make_willitem(valid=True)
+        item_b = _make_willitem(valid=True)
+
+        item_a.children = {}
+        item_b.children = {}
+
+        will = {"a": item_a, "b": item_b}
+
+        Will.set_invalidate("a", will)
+
+        assert item_a.get_status("INVALIDATED") is True
+        assert item_a.get_status("VALID") is False
+        assert item_b.get_status("INVALIDATED") is False
+        assert item_b.get_status("VALID") is True
+
+    def test_multiple_children(self):
+        """Invalidating a parent with multiple children cascades to all of them."""
+        parent = _make_willitem(valid=True)
+        child1 = _make_willitem(valid=True)
+        child2 = _make_willitem(valid=True)
+
+        parent.children = {
+            "c1": ["c1", 0, 0],
+            "c2": ["c2", 0, 0],
+        }
+        child1.children = {}
+        child2.children = {}
+
+        will = {"p": parent, "c1": child1, "c2": child2}
+
+        Will.set_invalidate("p", will)
+
+        assert parent.get_status("INVALIDATED") is True
+        assert child1.get_status("INVALIDATED") is True
+        assert child2.get_status("INVALIDATED") is True
+
+    def test_idempotent(self):
+        """Setting INVALIDATED twice on the same item is a safe no-op."""
+        item = _make_willitem(valid=True)
+        item.children = {}
+        will = {"wid": item}
+
+        Will.set_invalidate("wid", will)
+        Will.set_invalidate("wid", will)
+
+        assert item.get_status("INVALIDATED") is True
+        assert item.get_status("VALID") is False
+
+
+# ------------------------------------------------------------------ #
+# Main
+# ------------------------------------------------------------------ #
+
+if __name__ == "__main__":
+    for name in sorted(dir()):
+        if name.startswith("test_"):
+            globals()[name]()
+            print(f"  [OK] {name}")
+    print("[OK] All invalidation tests passed")