test: add python coin selection test

tests/test_spend.py

@@ -1,3 +1,5 @@
+import re
+
 from fixtures import *
 from test_framework.serializations import PSBT
 from test_framework.utils import wait_for, COIN, RpcError
@@ -219,3 +221,128 @@ def test_send_to_self(lianad, bitcoind):
         c for c in lianad.rpc.listcoins()["coins"] if c["spend_info"] is None
     )
     wait_for(lambda: len(list(unspent_coins())) == 1)
+
+
+def test_coin_selection(lianad, bitcoind):
+    """We can create a spend using coin selection."""
+    # Do not specify any coins in outpoints list in order to use coin selection.
+    outpoints = []
+    # Send to an (external) address.
+    dest_100_000 = {bitcoind.rpc.getnewaddress(): 100_000}
+    # Coin selection is not possible if we have no coins.
+    assert len(lianad.rpc.listcoins()["coins"]) == 0
+    with pytest.raises(
+        RpcError,
+        match=re.escape("Coin selection failed"),
+    ):
+        lianad.rpc.createspend(dest_100_000, outpoints, 2)
+
+    # Receive a coin in an unconfirmed deposit transaction.
+    recv_addr = lianad.rpc.getnewaddress()["address"]
+    deposit = bitcoind.rpc.sendtoaddress(recv_addr, 0.0008)  # 80_000 sats
+    wait_for(lambda: len(lianad.rpc.listcoins()["coins"]) == 1)
+    assert (
+        len(
+            [
+                c
+                for c in lianad.rpc.listcoins()["coins"]
+                if c["block_height"] is not None
+            ]
+        )
+        == 0
+    )
+    # There are still no coin selection candidates.
+    with pytest.raises(
+        RpcError,
+        match=re.escape("Coin selection failed"),
+    ):
+        lianad.rpc.createspend(dest_100_000, outpoints, 2)
+
+    # Confirm coin.
+    bitcoind.generate_block(1, wait_for_mempool=deposit)
+    confirmed_coins = lambda: (
+        c for c in lianad.rpc.listcoins()["coins"] if c["block_height"] is not None
+    )
+    wait_for(lambda: len(list(confirmed_coins())) == 1)
+
+    # Insufficient funds for coin selection.
+    with pytest.raises(
+        RpcError,
+        match=re.escape("Coin selection failed"),
+    ):
+        lianad.rpc.createspend(dest_100_000, outpoints, 2)
+
+    # Reduce spend amount.
+    dest_30_000 = {bitcoind.rpc.getnewaddress(): 30_000}
+    res = lianad.rpc.createspend(dest_30_000, outpoints, 2)
+    assert "psbt" in res
+
+    # The transaction must contain a change output.
+    spend_psbt = PSBT.from_base64(res["psbt"])
+    assert len(spend_psbt.o) == 2
+    assert len(spend_psbt.tx.vout) == 2
+
+    # Sign and broadcast this Spend transaction.
+    signed_psbt = lianad.signer.sign_psbt(spend_psbt)
+    lianad.rpc.updatespend(signed_psbt.to_base64())
+    spend_txid = signed_psbt.tx.txid().hex()
+    lianad.rpc.broadcastspend(spend_txid)
+
+    wait_for(lambda: len(lianad.rpc.listcoins()["coins"]) == 2)
+    coins = lianad.rpc.listcoins()["coins"]
+    # Check that change output is unconfirmed.
+    assert (
+        len(
+            [
+                c
+                for c in coins
+                if c["spend_info"] is None and c["block_height"] is not None
+            ]
+        )
+        == 0
+    )
+    # Check we cannot use coins as candidates if they have been spent or are unconfirmed.
+    with pytest.raises(
+        RpcError,
+        match=re.escape("Coin selection failed"),
+    ):
+        lianad.rpc.createspend(dest_30_000, outpoints, 2)
+
+    # Now confirm the Spend.
+    bitcoind.generate_block(1, wait_for_mempool=spend_txid)
+    wait_for(lambda: len(list(confirmed_coins())) == 2)
+
+    # We now have an unspent coin to use as candidate for coin selection.
+    assert (
+        len([c for c in lianad.rpc.listcoins()["coins"] if c["spend_info"] is None])
+        == 1
+    )
+    # But its value is not enough for this Spend.
+    dest_60_000 = {bitcoind.rpc.getnewaddress(): 60_000}
+    with pytest.raises(
+        RpcError,
+        match=re.escape("Coin selection failed"),
+    ):
+        lianad.rpc.createspend(dest_60_000, outpoints, 2)
+
+    # Get another coin to check coin selection with more than one candidate.
+    recv_addr = lianad.rpc.getnewaddress()["address"]
+    deposit = bitcoind.rpc.sendtoaddress(recv_addr, 0.0002)  # 20_000 sats
+    bitcoind.generate_block(1, wait_for_mempool=deposit)
+    wait_for(lambda: len(list(confirmed_coins())) == 3)
+
+    res = lianad.rpc.createspend(dest_60_000, outpoints, 2)
+    assert "psbt" in res
+
+    # The transaction must contain a change output.
+    spend_psbt = PSBT.from_base64(res["psbt"])
+    assert len(spend_psbt.o) == 2
+    assert len(spend_psbt.tx.vout) == 2
+
+    # Sign and broadcast this Spend transaction.
+    signed_psbt = lianad.signer.sign_psbt(spend_psbt)
+    lianad.rpc.updatespend(signed_psbt.to_base64())
+    spend_txid = signed_psbt.tx.txid().hex()
+    lianad.rpc.broadcastspend(spend_txid)
+    bitcoind.generate_block(1, wait_for_mempool=spend_txid)
+    wait_for(lambda: len(list(confirmed_coins())) == 4)