Test plugin improvements

contrib/pyln-testing/pyln/testing/inline-plugin.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python3
+"""Generic inline plugin shim: bridges lightningd stdio <-> inline-plugin.sock in cwd.
+Used by inline_plugin() in pyln/testing/utils.py."""
+import os
+import socket
+import sys
+import threading
+
+
+def _stdin_to_sock(conn):
+    while chunk := sys.stdin.buffer.read1(4096):
+        conn.sendall(chunk)
+    # Stdin closed means lightningd is done with us: exit immediately so the
+    # OS closes the socket and the serve thread can accept the next connection.
+    os._exit(0)
+
+
+conn = socket.socket(socket.AF_UNIX)
+conn.connect('inline-plugin.sock')
+
+threading.Thread(target=_stdin_to_sock, args=(conn,), daemon=True).start()
+
+while chunk := conn.recv(4096):
+    sys.stdout.buffer.write(chunk)
+    sys.stdout.buffer.flush()

contrib/pyln-testing/pyln/testing/utils.py

@@ -11,6 +11,7 @@
 from pyln.client import LightningRpc
 from pyln.client import Millisatoshi
 from pyln.client import NodeVersion
+from pyln.client import Plugin
 
 import ephemeral_port_reserve  # type: ignore
 import json
@@ -22,13 +23,15 @@
 import random
 import re
 import shutil
+import socket
 import sqlite3
 import string
 import struct
 import subprocess
 import sys
 import threading
 import time
+import types
 import warnings
 
 BITCOIND_CONFIG = {
@@ -78,6 +81,8 @@ def env(name, default=None):
 VALGRIND = env("VALGRIND") == "1"
 TEST_NETWORK = env("TEST_NETWORK", 'regtest')
 TEST_DEBUG = env("TEST_DEBUG", "0") == "1"
+
+INLINE_PLUGIN_PATH = os.path.join(os.path.dirname(__file__), 'inline-plugin.py')
 SLOW_MACHINE = env("SLOW_MACHINE", "0") == "1"
 DEPRECATED_APIS = env("DEPRECATED_APIS", "0") == "1"
 TIMEOUT = int(env("TIMEOUT", 180 if SLOW_MACHINE else 60))
@@ -1786,7 +1791,8 @@ def get_nodes(self, num_nodes, opts=None):
     def get_node(self, node_id=None, options=None, dbfile=None,
                  bkpr_dbfile=None, feerates=(15000, 11000, 7500, 3750),
                  start=True, wait_for_bitcoind_sync=True, may_fail=False,
-                 expect_fail=False, cleandir=True, gossip_store_file=None, unused_grpc_port=True, **kwargs):
+                 expect_fail=False, cleandir=True, gossip_store_file=None, unused_grpc_port=True,
+                 inline_plugin=None, **kwargs):
         node_id = self.get_node_id() if not node_id else node_id
         port = reserve_unused_port()
         grpc_port = self.get_unused_port() if unused_grpc_port else None
@@ -1830,6 +1836,11 @@ def get_node(self, node_id=None, options=None, dbfile=None,
             shutil.copy(gossip_store_file, os.path.join(node.daemon.lightning_dir, TEST_NETWORK,
                                                         'gossip_store'))
 
+        if inline_plugin is not None:
+            if 'plugin' not in node.daemon.opts:
+                node.daemon.opts['plugin'] = INLINE_PLUGIN_PATH
+            _inline_plugin(node, inline_plugin)
+
         if start:
             try:
                 node.start(wait_for_bitcoind_sync)
@@ -1944,3 +1955,64 @@ def killall(self, expected_successes):
             drop_unused_port(p)
 
         return not unexpected_fail, err_msgs
+
+
+def _inline_plugin(node, setup_fn):
+    """Set up an inline plugin serve thread for a not-yet-started node.
+
+    Normally called via get_node(inline_plugin=setup_fn).  The plugin's cwd
+    (set by lightningd) is node.daemon.lightning_dir/TEST_NETWORK/, which is
+    where the shim looks for inline-plugin.sock.
+
+    Example::
+
+        def setup(plugin):
+            @plugin.method('greet')
+            def greet(name, plugin):
+                return {'message': f'hello {name}'}
+
+        l1 = node_factory.get_node(inline_plugin=setup)
+        assert l1.rpc.greet('world') == {'message': 'hello world'}
+    """
+    sock_path = os.path.join(node.daemon.lightning_dir, TEST_NETWORK, 'inline-plugin.sock')
+    srv = socket.socket(socket.AF_UNIX)
+    srv.bind(sock_path)
+    srv.listen(1)
+
+    plugin = Plugin(autopatch=False)
+    setup_fn(plugin)
+
+    def serve():
+        while True:
+            conn, _ = srv.accept()
+
+            class _SockWriter:
+                def write(self, data):
+                    try:
+                        conn.sendall(data)
+                    except OSError:
+                        pass
+                def flush(self):
+                    pass
+
+            writer = _SockWriter()
+            plugin.stdout = types.SimpleNamespace(buffer=writer, flush=writer.flush)
+
+            partial = b""
+            while True:
+                try:
+                    chunk = conn.recv(4096)
+                except OSError:
+                    break
+                if not chunk:
+                    break
+                partial += chunk
+                msgs = partial.split(b'\n\n')
+                if len(msgs) < 2:
+                    continue
+                try:
+                    partial = plugin._multi_dispatch(msgs)
+                except Exception:
+                    break
+
+    threading.Thread(target=serve, daemon=True).start()