""" Mock tests for inheritance core logic. Tests: 1. prepare_transactions builds outputs matching heir amounts 2. prepare_transactions handles multiple locktime groups 3. new will item gets ANTICIPATED status via set_anticipate 4. REPLACED status propagates through the will tree 5. INVALIDATED status when input is spent and parent tx not tracked """ import sys, os, time, copy from unittest.mock import MagicMock, patch sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir)) from bal.core.heirs import ( Heirs, prepare_transactions, HEIR_ADDRESS, HEIR_REAL_AMOUNT, ) from bal.core.will import Will, WillItem from bal.core.util import Util from electrum.transaction import ( PartialTransaction, PartialTxInput, PartialTxOutput, TxOutpoint, ) from electrum.util import bfh ADDR1 = "1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa" ADDR2 = "1BvBMSEYstWetqTFn5Au4m4GFg7xJaNVN2" ADDR3 = "12c6DSiU4Rq3P4ZxziKxzrL5LmMBrzjrJX" ADDR_CHG = "3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy" TX_FEES = 100 def _utxo(txid_hex, value=500_000): txin = PartialTxInput(prevout=TxOutpoint(txid=bfh(txid_hex), out_idx=0)) txin._trusted_value_sats = value txin._TxInput__value_sats = value txin.is_mine = True return txin def _tx(utxos, outputs, locktime): return PartialTransaction.from_io(utxos, outputs, locktime=locktime, version=2) def _wi(tx, heirs_data): wi = WillItem.__new__(WillItem) wi.tx = tx wi._id = tx.txid() wi.heirs = heirs_data wi.we = None wi.status = "" wi.STATUS = copy.deepcopy(WillItem.STATUS_DEFAULT) wi.tx_fees = TX_FEES wi.father = None wi.children = {} wi.set_status("VALID", True) return wi def _txid_gen(): i = 0 while True: yield f"{i:064x}" i += 1 def _wallet(): w = MagicMock() w.dust_threshold.return_value = 500 w.get_change_addresses_for_new_transaction.return_value = [ADDR_CHG] w.get_utxos.return_value = [] w.db.get.return_value = {} w.db.get_transaction.return_value = None return w # ------------------------------------------------------------------ # # Test 1: prepare_transactions builds outputs matching heir amounts # ------------------------------------------------------------------ # @patch("electrum.transaction.bitcoin.address_to_script", return_value=b"\x00\x14" + b"\x00" * 20) @patch("bal.core.heirs.bitcoin.is_address", return_value=True) @patch("electrum.transaction.bitcoin.is_address", return_value=True) @patch("bal.core.heirs.PartialTransaction.txid", side_effect=_txid_gen()) def test_prepare_transactions_verify_amounts(*_): wallet = _wallet() now = int(time.time()) lt = now + 30 * 86400 locktimes = { lt: { "alice": [ADDR1, "fix", str(lt), 1_000_000], "bob": [ADDR2, "fix", str(lt), 1_000_000], "charlie": [ADDR3, "fix", str(lt), 1_000_000], } } utxos = [_utxo(f"{i:02x}" * 32, 500_000) for i in range(8)] fees = {lt: 10000} txs = prepare_transactions(locktimes, utxos[:], fees, wallet) assert txs, "no txs built" for txid, tx in txs.items(): outs = tx.outputs() for name, hdata in getattr(tx, "heirs", {}).items(): if "w!ll3x3c" in name: continue addr = hdata[HEIR_ADDRESS] exp = hdata[HEIR_REAL_AMOUNT] assert any(o.value == exp for o in outs), \ f"{name}: amount {exp} not in outputs" print(f"[OK] prepare_transactions built {len(txs)} tx(s)") # ------------------------------------------------------------------ # # Test 2: prepare_transactions with multiple locktime groups # ------------------------------------------------------------------ # @patch("electrum.transaction.bitcoin.address_to_script", return_value=b"\x00\x14" + b"\x00" * 20) @patch("bal.core.heirs.bitcoin.is_address", return_value=True) @patch("electrum.transaction.bitcoin.is_address", return_value=True) @patch("bal.core.heirs.PartialTransaction.txid", side_effect=_txid_gen()) def test_prepare_transactions_multiple_calls(*_): wallet = _wallet() now = int(time.time()) lt1, lt2 = now + 30 * 86400, now + 60 * 86400 utxos_pool = [_utxo(f"{i:02x}" * 32, 600_000) for i in range(2)] # First call: process lt1 group txs1 = prepare_transactions( {lt1: {"alice": [ADDR1, 500_000, str(lt1)]}}, utxos_pool[:], {lt1: 5000}, wallet, ) assert txs1, "no txs from first group" t1_heirs = set() for tx in txs1.values(): t1_heirs.update(getattr(tx, "heirs", {}).keys()) assert "alice" in t1_heirs, "alice missing from first call" # Second call: process lt2 group (with fresh mock data) txs2 = prepare_transactions( {lt2: {"bob": [ADDR2, 300_000, str(lt2)]}}, [_utxo("ff" * 32, 600_000)], {lt2: 5000}, wallet, ) assert txs2, "no txs from second group" t2_heirs = set() for tx in txs2.values(): t2_heirs.update(getattr(tx, "heirs", {}).keys()) assert "bob" in t2_heirs, "bob missing from second call" print("[OK] prepare_transactions handles multiple sequential calls") # ------------------------------------------------------------------ # # Test 3: set_anticipate gives ANTICIPATED status + lower locktime # ------------------------------------------------------------------ # def test_set_anticipate(): now = int(time.time()) lt = int((now + 30 * 86400) / 86400) * 86400 old_tx = _tx([_utxo("aa" * 32, 200_000)], [PartialTxOutput.from_address_and_value(ADDR1, 100_000)], lt) old_wi = _wi(old_tx, {"alice": [ADDR1, 100_000, str(lt)]}) new_tx = _tx([_utxo("bb" * 32, 200_000)], [PartialTxOutput.from_address_and_value(ADDR2, 100_000)], lt) new_wi = _wi(new_tx, {"alice": [ADDR2, 100_000, str(lt)]}) anticipated = new_wi.set_anticipate(old_wi) assert anticipated, "set_anticipate should return True" assert new_wi.get_status("ANTICIPATED"), "ANTICIPATED should be set" assert new_wi.get_status("VALID"), "VALID should remain True" assert new_wi.tx.locktime <= old_wi.tx.locktime, \ f"new locktime {new_wi.tx.locktime} should be <= old {old_wi.tx.locktime}" print("[OK] set_anticipate sets ANTICIPATED, keeps VALID, lowers locktime") # ------------------------------------------------------------------ # # Test 4: REPLACED propagates through will tree # ------------------------------------------------------------------ # def test_replaced_propagates(): now = int(time.time()) lt_parent = int(now + 60 * 86400) lt_child = int(now + 30 * 86400) lt_replacer = int(now + 15 * 86400) # Pre-determine parent txid (can't use real tx.txid() on unsigned PSBT) parent_txid_hex = "aa" * 32 # Build parent tx spending the wallet utxo parent_utxo = _utxo("aa" * 32, 200_000) parent_tx = _tx([parent_utxo], [PartialTxOutput.from_address_and_value(ADDR1, 100_000), PartialTxOutput.from_address_and_value(ADDR_CHG, 99_000)], lt_parent) # Build child tx spending the parent's change output (parent_txid:1) ci = PartialTxInput(prevout=TxOutpoint(txid=bfh(parent_txid_hex), out_idx=1)) ci._trusted_value_sats = 99_000; ci.is_mine = True child_tx = _tx([ci], [PartialTxOutput.from_address_and_value(ADDR2, 50_000)], lt_child) # Build replacer tx spending the SAME utxo as parent, with lower locktime replacer_tx = _tx([_utxo("aa" * 32, 200_000)], [PartialTxOutput.from_address_and_value(ADDR3, 150_000)], lt_replacer) hp = {"a": [ADDR1, 100_000, str(lt_parent)]} hc = {"b": [ADDR2, 50_000, str(lt_child)]} hr = {"c": [ADDR3, 150_000, str(lt_replacer)]} pwi = _wi(parent_tx, hp) pwi._id = parent_txid_hex cwi = _wi(child_tx, hc) cwi._id = "bb" * 32 rwi = _wi(replacer_tx, hr) rwi._id = "cc" * 32 will = {pwi._id: pwi, cwi._id: cwi, rwi._id: rwi} Will.add_willtree(will) all_utxos = [_utxo("aa" * 32, 200_000)] Will.search_rai(Will.get_all_inputs(will, only_valid=True), all_utxos, will, _wallet()) assert pwi.get_status("REPLACED"), "parent should be REPLACED" assert not pwi.get_status("VALID"), "parent should NOT be VALID" assert cwi.get_status("REPLACED"), "child should be REPLACED (propagated)" assert not cwi.get_status("VALID"), "child should NOT be VALID" assert rwi.get_status("VALID"), "replacer should be VALID" assert not rwi.get_status("REPLACED"), "replacer should NOT be REPLACED" print("[OK] REPLACED propagation to children verified") # ------------------------------------------------------------------ # # Test 5: INVALIDATED when input not in utxos and parent TX untracked # ------------------------------------------------------------------ # def test_invalidated(): now = int(time.time()) lt = int(now + 30 * 86400) addr = ADDR1 # TX that spends an unknown input (not in wallet utxos) txin = PartialTxInput(prevout=TxOutpoint(txid=bfh("dd" * 32), out_idx=0)) txin._trusted_value_sats = 100_000; txin.is_mine = True tx = _tx([txin], [PartialTxOutput.from_address_and_value(addr, 90_000)], lt) wi = _wi(tx, {"a": [addr, 90_000, str(lt)]}) will = {wi._id: wi} Will.add_willtree(will) # all_utxos is empty, wallet.db.get_transaction returns None Will.search_rai(Will.get_all_inputs(will, only_valid=True), [], will, _wallet()) assert wi.get_status("INVALIDATED"), "should be INVALIDATED" assert not wi.get_status("VALID"), "should NOT be VALID" print("[OK] INVALIDATED when input not ours and parent TX untracked") # ------------------------------------------------------------------ # if __name__ == "__main__": import logging logging.basicConfig(level=logging.WARNING) for name in sorted(dir()): if name.startswith("test_"): globals()[name]() print("[OK] All mock inheritance tests passed")