pytest: simple test for pay's custom notifications.
What changed, and why it matters
This commit adds a new automated test to the Core Lightning project. It checks that the 'pay' command correctly sends three types of plugin notifications (channel hint updates, payment success, and payment failure). There is no change to production code, no bug fix, and no security-related content.
No security action needed. This is a regression test addition; review as normal code quality/test coverage change.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The diff is a single new pytest function, test_pay_plugin_notifications, in tests/test_plugin.py. It loads a test plugin (all_notifications.py), creates a three-node line graph, pays an invoice, and verifies that channel_hint_update, pay_success, and pay_failure notifications are emitted with expected payloads. It also imports first_scid from utils. No application logic is modified.
Changed components
tests/test_plugin.pyInspect captured patch +65 / −1
diff --git a/tests/test_plugin.py b/tests/test_plugin.py
index d1c6f293..9abe7c60 100644
--- a/tests/test_plugin.py
+++ b/tests/test_plugin.py
@@ -10,7 +10,7 @@ from utils import (
expected_peer_features, expected_node_features,
expected_channel_features, account_balance,
check_coin_moves, first_channel_id, EXPERIMENTAL_DUAL_FUND,
- mine_funding_to_announce, VALGRIND
+ mine_funding_to_announce, VALGRIND, first_scid
)
import ast
@@ -4374,6 +4374,70 @@ def test_peer_storage(node_factory, bitcoind):
assert not l2.daemon.is_in_log(r'PeerStorageFailed')
+def test_pay_plugin_notifications(node_factory, bitcoind, chainparams):
+ plugin = os.path.join(os.getcwd(), 'tests/plugins/all_notifications.py')
+ opts = {"plugin": plugin}
+
+ l1, l2, l3 = node_factory.line_graph(3, opts=[opts, {}, {}],
+ wait_for_announce=True)
+
+ def zero_timestamps(obj):
+ if isinstance(obj, dict):
+ for k, v in obj.items():
+ if k == "timestamp":
+ obj[k] = 0
+ else:
+ zero_timestamps(v)
+ elif isinstance(obj, list):
+ for item in obj:
+ zero_timestamps(item)
+ # other types are ignored
+ return obj
+
+ inv1 = l3.rpc.invoice(20000, "first", "desc")
+ l1.rpc.pay(inv1['bolt11'])
+
+ # It gets a channel hint update notification
+ line = l1.daemon.wait_for_log(f"plugin-all_notifications.py: notification channel_hint_update: ")
+ dict_str = line.split("notification channel_hint_update: ", 1)[1]
+ data = zero_timestamps(ast.literal_eval(dict_str))
+
+ # pyln-client's plugin.py duplicated payload into same name as update.
+ channel_hint_update_core = {'scid': first_scid(l1, l2) + '/1',
+ 'estimated_capacity_msat': 964719000 if chainparams['elements'] else 978718000,
+ 'total_capacity_msat': 1000000000,
+ 'timestamp': 0,
+ 'enabled': True}
+ channel_hint_update = {'origin': 'pay',
+ 'payload': {'channel_hint': channel_hint_update_core}}
+ assert data == channel_hint_update
+
+ # It gets a success notification
+ line = l1.daemon.wait_for_log(f"plugin-all_notifications.py: notification pay_success: ")
+ dict_str = line.split("notification pay_success: ", 1)[1]
+ data = ast.literal_eval(dict_str)
+ success_core = {'payment_hash': inv1['payment_hash'],
+ 'bolt11': inv1['bolt11']}
+ # Includes deprecated and modern. pyln-client plugin.py copies fields as necessary.
+ success = {'origin': 'pay',
+ 'payload': success_core}
+ assert data == success
+
+ inv2 = l3.rpc.invoice(10000, "second", "desc")
+ l3.rpc.delinvoice('second', 'unpaid')
+ with pytest.raises(RpcError, match="WIRE_INCORRECT_OR_UNKNOWN_PAYMENT_DETAILS"):
+ l1.rpc.pay(inv2['bolt11'])
+
+ line = l1.daemon.wait_for_log(f"plugin-all_notifications.py: notification pay_failure: ")
+ dict_str = line.split("notification pay_failure: ", 1)[1]
+ data = ast.literal_eval(dict_str)
+ failure_core = {'payment_hash': inv2['payment_hash'], 'bolt11': inv2['bolt11'], 'error': {'message': 'failed: WIRE_INCORRECT_OR_UNKNOWN_PAYMENT_DETAILS (reply from remote)'}}
+ # Includes deprecated and modern.
+ failure = {'origin': 'pay',
+ 'payload': failure_core}
+ assert data == failure
+
+
@pytest.mark.openchannel('v1')
@pytest.mark.openchannel('v2')
def test_openchannel_hook_channel_type(node_factory, bitcoind):
Why this scored 15/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.