""" Test: postponing a signed will requests invalidation of the old transaction. Scenario: The user has a signed/pushed will with RELATIVE locktime "300d". After 2 days they change the locktime to "1y" (365d, longer → postpone). The check flow must raise ``WillPostponedException`` because an already-sent will with an earlier locktime must be invalidated on-chain FIRST (otherwise a will-executor could broadcast the old tx and execute the inheritance too early). At the same time, the threshold check (``CheckAliveError``) must NOT fire because the threshold is the relative string "30d", which resolves against ``datetime.now()`` → always future. """ import copy import os import sys import time import itertools import pytest from unittest.mock import MagicMock, patch from datetime import datetime from electrum import constants sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) @pytest.fixture(autouse=True) def _regtest_network(): 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 Heirs # noqa: E402 from bal.core.will import ( # noqa: E402 Will, WillItem, WillPostponedException, ) from bal.core.willexecutors import Willexecutors # noqa: E402 from bal.core.plugin_base import BalTimestamp # noqa: E402 from electrum.transaction import PartialTransaction # noqa: E402 DAY = 86400 # ------------------------------------------------------------------ # # Helpers # ------------------------------------------------------------------ # def _willexecutors(specs): 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 _build(heirs, willexecutors, *, from_locktime=None, tx_fees=100): wallet = fx.fake_wallet() utxos = fx.make_utxos(4, 5_000_000) plugin = fx.fake_bal_plugin(willexecutors) if from_locktime is None: from_locktime = int(time.time()) counter = itertools.count(1) def fake_txid(self): 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 _tx_for_we(txs, url): for tx in txs.values(): we = getattr(tx, "willexecutor", None) if we and we.get("url") == url: return tx return None # ------------------------------------------------------------------ # # The test # ------------------------------------------------------------------ # def test_postpone_after_two_days_requests_invalidation(): """Postponing a signed will after 2 days raises WillPostponedException.""" now = int(time.time()) ha = fx.heir_addresses(3) we_addr = fx.willexecutor_addresses(1)[0] # ── 1. Create heirs with INITIAL locktime "300d" ───────────────── heirs_initial = { "alice": [ha[0], "40%", "300d"], "bob": [ha[1], "30%", "300d"], "carol": [ha[2], "30%", "300d"], } h_initial = Heirs.__new__(Heirs) dict.__init__(h_initial) h_initial.db = MagicMock() h_initial.wallet = MagicMock() h_initial.update(copy.deepcopy(heirs_initial)) wes = _willexecutors([("https://we.example", we_addr, 100000)]) # ── 2. Build the will ─────────────────────────────────────────── txs = _build(h_initial, wes, from_locktime=now) tx = _tx_for_we(txs, "https://we.example") assert tx is not None, "no will-executor transaction built" txid = tx._test_txid if hasattr(tx, "_test_txid") else tx.txid() # ── 3. Wrap in WillItem, mark SIGNED + PUSHED ─────────────────── 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 = now item.father = None item.children = {} item.description = "" item.change = "" item.set_status("VALID", True) item.set_status("COMPLETE", True) # ← signed item.set_status("PUSHED", True) # ← sent to will-executor willitems = {txid: item} old_locktime = int(tx.locktime) # ── 4. Advance "now" by 2 days ────────────────────────────────── advanced_now_dt = datetime.fromtimestamp(now + 2 * DAY) # Patch BalTimestamp's datetime.now() for threshold resolution. import bal.core.plugin_base as pb_mod import datetime as real_dt class _PatchedDt: @classmethod def now(cls): return advanced_now_dt @classmethod def fromtimestamp(cls, ts): return real_dt.datetime.fromtimestamp(ts) @classmethod def today(cls): return advanced_now_dt.date() _PatchedDt.timedelta = real_dt.timedelta with patch.object(pb_mod, "datetime", _PatchedDt): check_date_2d = BalTimestamp("30d").to_timestamp() # ── 5. Threshold still in future → no CheckAliveError ─────────── assert check_date_2d > advanced_now_dt.timestamp(), ( "30d threshold is still future after 2 days" ) # ── 6. check_will_expired — old locktime (now+300d) > check_date ─ all_inputs = Will.get_all_inputs(willitems, only_valid=True) all_inputs_min = Will.get_all_inputs_min_locktime(all_inputs) Will.check_will_expired(all_inputs_min, check_date_2d) # ── 7. check_willexecutors_and_heirs with CHANGED locktime ────── # Change the locktime from "300d" to "1y" (postpone). heirs_changed = { "alice": [ha[0], "40%", "1y"], "bob": [ha[1], "30%", "1y"], "carol": [ha[2], "30%", "1y"], } h_changed = Heirs.__new__(Heirs) dict.__init__(h_changed) h_changed.db = MagicMock() h_changed.wallet = MagicMock() h_changed.update(copy.deepcopy(heirs_changed)) with pytest.raises(WillPostponedException) as exc_info: Will.check_willexecutors_and_heirs( willitems, h_changed, wes, False, # no_willexecutor check_date_2d, 100, # tx_fees (unchanged) ) msg = str(exc_info.value) assert "postponed" in msg, f"exception should indicate postponement: {msg}" assert "300d" in msg, ( f"exception should mention the old locktime spec: {msg}" ) assert "1y" in msg, ( f"exception should mention the new locktime spec: {msg}" )