Skip to content

Add python integration test for spontaneous keysend payments #923

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
heyolaniran wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Add python integration test for spontaneous keysend payments #923

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5e3396e
Introduce helpers for TestLdkNode
heyolaniran Jun 15, 2026
3c9be11
add test for spontaneous payment
heyolaniran Jun 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
152 changes: 103 additions & 49 deletions bindings/python/src/ldk_node/test_ldk_node.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import os
import re
import requests
import socket

from ldk_node import *

Expand Down Expand Up @@ -118,8 +119,68 @@ def expect_event(node, expected_event_type):
assert isinstance(event, expected_event_type)
print("EVENT:", event)
node.event_handled()
return event

return event

def find_two_free_ports():
with socket.socket() as s1, socket.socket() as s2:
s1.bind(("127.0.0.1", 0))
s2.bind(("127.0.0.1",0))
port_1 = s1.getsockname()[1]
port_2 = s2.getsockname()[1]
return port_1, port_2

def setup_two_nodes(esplora_endpoint):
port_1, port_2 = find_two_free_ports()
tmp_dir_1 = tempfile.TemporaryDirectory("_ldk_node_1")
listening_addresses_1 = [f"127.0.0.1:{port_1}"]
node_1 = setup_node(tmp_dir_1.name, esplora_endpoint, listening_addresses_1)
node_1.start()
node_id_1 = node_1.node_id()

tmp_dir_2 = tempfile.TemporaryDirectory("_ldk_node_2")
listening_addresses_2 = [f"127.0.0.1:{port_2}"]
node_2 = setup_node(tmp_dir_2.name, esplora_endpoint, listening_addresses_2)
node_2.start()
node_id_2 = node_2.node_id()

return node_1, node_2, tmp_dir_1, tmp_dir_2, node_id_1, node_id_2, listening_addresses_2

def fund_nodes(node_1, node_2, esplora_endpoint, amount_sats=100000):
Comment thread
heyolaniran marked this conversation as resolved.
address_1 = node_1.onchain_payment().new_address()
txid_1 = send_to_address(address_1, amount_sats)
address_2 = node_2.onchain_payment().new_address()
txid_2 = send_to_address(address_2, amount_sats)

wait_for_tx(esplora_endpoint, txid_1)
wait_for_tx(esplora_endpoint, txid_2)
mine_and_wait(esplora_endpoint, 6)

node_1.sync_wallets()
node_2.sync_wallets()

def open_channel_and_wait_ready(node_1, node_2, node_id_2, listening_address_2, esplora_endpoint, channel_amount_sats=50000):
node_1.open_channel(node_id_2, listening_address_2, channel_amount_sats, None, None)

channel_pending_event_1 = expect_event(node_1, Event.CHANNEL_PENDING)
expect_event(node_2, Event.CHANNEL_PENDING)

funding_txid = channel_pending_event_1.funding_txo.txid
wait_for_tx(esplora_endpoint, funding_txid)
mine_and_wait(esplora_endpoint, 6)

node_1.sync_wallets()
node_2.sync_wallets()

channel_ready_event_1 = expect_event(node_1, Event.CHANNEL_READY)
channel_ready_event_2 = expect_event(node_2, Event.CHANNEL_READY)
return channel_ready_event_1, channel_ready_event_2, funding_txid

def stop_and_cleanup(node_1, node_2, tmp_dir_1, tmp_dir_2):
node_1.stop()
node_2.stop()
time.sleep(1)
tmp_dir_1.cleanup()
tmp_dir_2.cleanup()


class TestLdkNode(unittest.TestCase):
Expand All @@ -130,41 +191,53 @@ def setUp(self):
esplora_endpoint = get_esplora_endpoint()
mine_and_wait(esplora_endpoint, 1)

def test_channel_full_cycle(self):
def test_spontaneous_payment(self):
"""Spontaneous payment test in python: keysend after channel ready."""
esplora_endpoint = get_esplora_endpoint()

## Setup Node 1
tmp_dir_1 = tempfile.TemporaryDirectory("_ldk_node_1")
print("TMP DIR 1:", tmp_dir_1.name)
node_1, node_2, tmp_dir_1, tmp_dir_2, node_id_1, node_id_2, listening_addresses_2 = setup_two_nodes(esplora_endpoint)
fund_nodes(node_1, node_2, esplora_endpoint)
open_channel_and_wait_ready(node_1, node_2, node_id_2, listening_addresses_2[0], esplora_endpoint)

listening_addresses_1 = ["127.0.0.1:2323"]
node_1 = setup_node(tmp_dir_1.name, esplora_endpoint, listening_addresses_1)
node_1.start()
node_id_1 = node_1.node_id()
print("Node ID 1:", node_id_1)
keysend_amount_msat = 2_500_000
custom_tlvs = [CustomTlvRecord(type_num=13377331, value=bytes([1, 2, 3]))]
keysend_payment_id = node_1.spontaneous_payment().send_with_custom_tlvs(
keysend_amount_msat, node_id_2, None, custom_tlvs
)

# Setup Node 2
tmp_dir_2 = tempfile.TemporaryDirectory("_ldk_node_2")
print("TMP DIR 2:", tmp_dir_2.name)
expect_event(node_1, Event.PAYMENT_SUCCESSFUL)
received_event = expect_event(node_2, Event.PAYMENT_RECEIVED)

listening_addresses_2 = ["127.0.0.1:2324"]
node_2 = setup_node(tmp_dir_2.name, esplora_endpoint, listening_addresses_2)
node_2.start()
node_id_2 = node_2.node_id()
print("Node ID 2:", node_id_2)
self.assertEqual(received_event.amount_msat, keysend_amount_msat)
self.assertEqual(received_event.custom_records, custom_tlvs)

address_1 = node_1.onchain_payment().new_address()
txid_1 = send_to_address(address_1, 100000)
address_2 = node_2.onchain_payment().new_address()
txid_2 = send_to_address(address_2, 100000)
sender_payment = node_1.payment(keysend_payment_id)
receiver_payment = node_2.payment(keysend_payment_id)

wait_for_tx(esplora_endpoint, txid_1)
wait_for_tx(esplora_endpoint, txid_2)
self.assertIsNotNone(sender_payment)
self.assertIsNotNone(receiver_payment)
self.assertEqual(sender_payment.status, PaymentStatus.SUCCEEDED)
self.assertEqual(sender_payment.direction, PaymentDirection.OUTBOUND)
self.assertEqual(sender_payment.amount_msat, keysend_amount_msat)
self.assertTrue(sender_payment.kind.is_spontaneous())

mine_and_wait(esplora_endpoint, 6)
self.assertEqual(receiver_payment.status, PaymentStatus.SUCCEEDED)
self.assertEqual(receiver_payment.direction, PaymentDirection.INBOUND)
self.assertEqual(receiver_payment.amount_msat, keysend_amount_msat)
self.assertTrue(receiver_payment.kind.is_spontaneous())

stop_and_cleanup(node_1, node_2, tmp_dir_1, tmp_dir_2)

def test_channel_full_cycle(self):
esplora_endpoint = get_esplora_endpoint()

## Setup two nodes
node_1, node_2, tmp_dir_1, tmp_dir_2, node_id_1, node_id_2, listening_addresses_2 = setup_two_nodes(esplora_endpoint)
print("Node ID 1:", node_id_1)
print("Node ID 2:", node_id_2)

fund_nodes(node_1, node_2, esplora_endpoint)

node_1.sync_wallets()
node_2.sync_wallets()

spendable_balance_1 = node_1.list_balances().spendable_onchain_balance_sats
spendable_balance_2 = node_2.list_balances().spendable_onchain_balance_sats
Expand All @@ -183,22 +256,9 @@ def test_channel_full_cycle(self):
print("TOTAL 2:", total_balance_2)
self.assertEqual(total_balance_2, 100000)

node_1.open_channel(node_id_2, listening_addresses_2[0], 50000, None, None)


channel_pending_event_1 = expect_event(node_1, Event.CHANNEL_PENDING)
channel_pending_event_2 = expect_event(node_2, Event.CHANNEL_PENDING)
funding_txid = channel_pending_event_1.funding_txo.txid
wait_for_tx(esplora_endpoint, funding_txid)
mine_and_wait(esplora_endpoint, 6)

node_1.sync_wallets()
node_2.sync_wallets()

channel_ready_event_1 = expect_event(node_1, Event.CHANNEL_READY)
channel_ready_event_1, channel_ready_event_2, funding_txid = open_channel_and_wait_ready(node_1, node_2, node_id_2, listening_addresses_2[0], esplora_endpoint)
print("funding_txo:", funding_txid)

channel_ready_event_2 = expect_event(node_2, Event.CHANNEL_READY)

description = Bolt11InvoiceDescription.DIRECT("asdf")
invoice = node_2.bolt11_payment().receive(2500000, description, 9217)
Expand Down Expand Up @@ -228,13 +288,7 @@ def test_channel_full_cycle(self):
self.assertEqual(spendable_balance_after_close_2, 102500)

# Stop nodes
node_1.stop()
node_2.stop()

# Cleanup
time.sleep(1) # Wait a sec so our logs can finish writing
tmp_dir_1.cleanup()
tmp_dir_2.cleanup()
stop_and_cleanup(node_1, node_2, tmp_dir_1, tmp_dir_2)

if __name__ == '__main__':
unittest.main()