pytest: convert many tests to use inline_plugin helper.

diff --git a/tests/plugins/block_added.py b/tests/plugins/block_added.py
deleted file mode 100755
index 9da46587..00000000
--- a/tests/plugins/block_added.py
+++ /dev/null
@@ -1,21 +0,0 @@
-#!/usr/bin/env python3
-
-from pyln.client import Plugin
-
-
-plugin = Plugin()
-
-blocks_catched = []
-
-
-@plugin.subscribe("block_added")
-def notify_block_added(plugin, block_added, **kwargs):
-    blocks_catched.append(block_added["height"])
-
-
-@plugin.method("blockscatched")
-def return_moves(plugin):
-    return blocks_catched
-
-
-plugin.run()
diff --git a/tests/plugins/custom_notifications.py b/tests/plugins/custom_notifications.py
deleted file mode 100755
index 1a3d92f1..00000000
--- a/tests/plugins/custom_notifications.py
+++ /dev/null
@@ -1,55 +0,0 @@
-#!/usr/bin/env python3
-from pyln.client import Plugin
-
-
-plugin = Plugin()
-
-
-@plugin.subscribe("custom")
-def on_custom_notification(origin, message, **kwargs):
-    plugin.log("Got a custom notification {} from plugin {}".format(message, origin))
-
-
-@plugin.method("emit")
-def emit(plugin):
-    """Emit a simple string notification to topic "custom"
-    """
-    plugin.notify("custom", {'message': "Hello world"})
-
-
-@plugin.method("faulty-emit")
-def faulty_emit(plugin):
-    """Emit a simple string notification to topic "custom"
-    """
-    plugin.notify("ididntannouncethis", {'message': "Hello world"})
-
-
-@plugin.subscribe("pay_success")
-def on_pay_success(origin, pay_success, **kwargs):
-    plugin.log(
-        "Got a pay_success notification from plugin {} for payment_hash {}".format(
-            origin,
-            pay_success['payment_hash']
-        )
-    )
-
-
-@plugin.subscribe("pay_part_start")
-def on_pay_part_start(origin, **kwargs):
-    plugin.log("Got pay_part_start: {}".format(kwargs))
-
-
-@plugin.subscribe("pay_part_end")
-def on_pay_part_end(origin, **kwargs):
-    plugin.log("Got pay_part_end: {}".format(kwargs))
-
-
-@plugin.subscribe("ididntannouncethis")
-def on_faulty_emit(origin, payload, **kwargs):
-    """We should never receive this as it gets dropped.
-    """
-    plugin.log("Got the ididntannouncethis event")
-
-
-plugin.add_notification_topic("custom")
-plugin.run()
diff --git a/tests/plugins/fail_htlcs_invalid.py b/tests/plugins/fail_htlcs_invalid.py
deleted file mode 100755
index 95881d8b..00000000
--- a/tests/plugins/fail_htlcs_invalid.py
+++ /dev/null
@@ -1,20 +0,0 @@
-#!/usr/bin/env python3
-
-from pyln.client import Plugin
-
-plugin = Plugin()
-
-
-@plugin.hook("htlc_accepted")
-def on_htlc_accepted(onion, plugin, **kwargs):
-    plugin.log("Failing htlc on purpose with invalid onion failure")
-    plugin.log("onion: %r" % (onion))
-    # WIRE_TEMPORARY_CHANNEL_FAILURE = 0x1007
-    # This failure code should be followed by a
-    # `channel_update`; we deliberately return
-    # a 0-length `channel_update` to trigger
-    # issue #3757 reported by @sumBTC.
-    return {"result": "fail", "failure_message": "10070000"}
-
-
-plugin.run()
diff --git a/tests/plugins/multiline-help.py b/tests/plugins/multiline-help.py
deleted file mode 100755
index 4b8b3a7f..00000000
--- a/tests/plugins/multiline-help.py
+++ /dev/null
@@ -1,17 +0,0 @@
-#!/usr/bin/env python3
-from pyln.client import Plugin, Millisatoshi
-
-
-plugin = Plugin()
-
-
-@plugin.method("helpme")
-def helpme(plugin, msat: Millisatoshi):
-    """This is a message which consumes multiple lines and thus should
-    be well-formatted by lightning-cli help
-
-    """
-    return {'help': msat}
-
-
-plugin.run()
diff --git a/tests/plugins/onionmessage_forward_fail_notification.py b/tests/plugins/onionmessage_forward_fail_notification.py
deleted file mode 100755
index 69606e23..00000000
--- a/tests/plugins/onionmessage_forward_fail_notification.py
+++ /dev/null
@@ -1,20 +0,0 @@
-#!/usr/bin/env python3
-
-""" We get an onionmessage_forward_fail notification, and open a connection
-"""
-from pyln.client import Plugin
-
-plugin = Plugin()
-
-
-@plugin.subscribe("onionmessage_forward_fail")
-def on_onionmessage_forward_fail(onionmessage_forward_fail, **kwargs):
-    plugin.log(f"Received onionmessage_forward_fail {onionmessage_forward_fail}")
-
-    plugin.rpc.connect(onionmessage_forward_fail['next_node_id'])
-    # injectonionmessage expects to unwrap, so hand it *incoming*
-    plugin.rpc.injectonionmessage(onionmessage_forward_fail['path_key'],
-                                  onionmessage_forward_fail['incoming'])
-
-
-plugin.run()
diff --git a/tests/plugins/openchannel_hook_delay.py b/tests/plugins/openchannel_hook_delay.py
deleted file mode 100755
index 09ec2abd..00000000
--- a/tests/plugins/openchannel_hook_delay.py
+++ /dev/null
@@ -1,32 +0,0 @@
-#!/usr/bin/env python3
-"""Plugin to test openchannel_hook
-
-Will simply accept any channel. Useful fot testing chained hook.
-"""
-
-from pyln.client import Plugin
-import time
-
-plugin = Plugin()
-
-
-@plugin.hook('openchannel')
-def on_openchannel(openchannel, plugin, **kwargs):
-    delaytime = float(plugin.get_option('delaytime'))
-    msg = f'delaying WIRE_ACCEPT_CHANNEL for {delaytime}s'
-    plugin.log(msg)
-    time.sleep(delaytime)
-    return {'result': 'continue'}
-
-
-@plugin.hook('openchannel2')
-def on_openchannel2(openchannel2, plugin, **kwargs):
-    delaytime = float(plugin.get_option('delaytime'))
-    msg = f'delaying WIRE_ACCEPT_CHANNEL for {delaytime}s'
-    plugin.log(msg)
-    time.sleep(delaytime)
-    return {'result': 'continue'}
-
-
-plugin.add_option('delaytime', '10', 'How long to hold the WIRE_OPEN_CHANNEL.')
-plugin.run()
diff --git a/tests/plugins/pretend_badlog.py b/tests/plugins/pretend_badlog.py
deleted file mode 100755
index a255fe11..00000000
--- a/tests/plugins/pretend_badlog.py
+++ /dev/null
@@ -1,32 +0,0 @@
-#!/usr/bin/env python3
-"""This plugin is used to check that warning(unusual/broken level log) calls are working correctly.
-"""
-from pyln.client import Plugin
-
-plugin = Plugin()
-
-
-@plugin.init()
-def init(configuration, options, plugin):
-    plugin.log("initialized")
-
-
-@plugin.subscribe("warning")
-def notify_warning(plugin, warning, **kwargs):
-    plugin.log("Received warning")
-    plugin.log("level: {}".format(warning['level']))
-    plugin.log("time: {}".format(warning['time']))
-    plugin.log("source: {}".format(warning['source']))
-    plugin.log("log: {}".format(warning['log']))
-
-
-@plugin.method("pretendbad")
-def pretend_bad(event, level, plugin):
-    """Log an specified level entry.
-    And in plugin, we use 'warn'/'error' instead of
-    'unusual'/'broken'
-    """
-    plugin.log("{}".format(event), level)
-
-
-plugin.run()
diff --git a/tests/plugins/print_htlc_onion.py b/tests/plugins/print_htlc_onion.py
deleted file mode 100755
index f1557dd2..00000000
--- a/tests/plugins/print_htlc_onion.py
+++ /dev/null
@@ -1,19 +0,0 @@
-#!/usr/bin/env python3
-"""Plugin that prints out HTLC onions.
-
-We use this to check whether they're TLV or not
-
-"""
-
-from pyln.client import Plugin
-
-plugin = Plugin()
-
-
-@plugin.hook("htlc_accepted")
-def on_htlc_accepted(htlc, onion, plugin, **kwargs):
-    plugin.log("Got onion {}".format(onion))
-    return {'result': 'continue'}
-
-
-plugin.run()
diff --git a/tests/plugins/reject_odd_funding_amounts.py b/tests/plugins/reject_odd_funding_amounts.py
deleted file mode 100755
index fc4d67a3..00000000
--- a/tests/plugins/reject_odd_funding_amounts.py
+++ /dev/null
@@ -1,36 +0,0 @@
-#!/usr/bin/env python3
-"""Simple plugin to test the openchannel_hook.
-
-We just refuse to let them open channels with an odd amount of millisatoshis.
-"""
-
-from pyln.client import Plugin, Millisatoshi
-
-plugin = Plugin()
-
-
-def run_check(funding_amt_str):
-    if Millisatoshi(funding_amt_str).to_satoshi() % 2 == 1:
-        return {'result': 'reject', 'error_message': "I don't like odd amounts"}
-
-    return {'result': 'continue'}
-
-
-@plugin.hook('openchannel')
-def on_openchannel(openchannel, plugin, **kwargs):
-    print("{} VARS".format(len(openchannel.keys())))
-    for k in sorted(openchannel.keys()):
-        print("{}={}".format(k, openchannel[k]))
-    return run_check(openchannel['funding_msat'])
-
-
-@plugin.hook('openchannel2')
-def on_openchannel2(openchannel2, plugin, **kwargs):
-    print("{} VARS".format(len(openchannel2.keys())))
-    for k in sorted(openchannel2.keys()):
-        print("{}={}".format(k, openchannel2[k]))
-
-    return run_check(openchannel2['their_funding_msat'])
-
-
-plugin.run()
diff --git a/tests/plugins/reject_some_invoices.py b/tests/plugins/reject_some_invoices.py
deleted file mode 100755
index 321e4e7a..00000000
--- a/tests/plugins/reject_some_invoices.py
+++ /dev/null
@@ -1,25 +0,0 @@
-#!/usr/bin/env python3
-"""Simple plugin to test the invoice_payment_hook.
-
-We just refuse to let them pay invoices with preimages divisible by 16.
-"""
-
-from pyln.client import Plugin
-
-plugin = Plugin()
-
-
-@plugin.hook('invoice_payment')
-def on_payment(payment, plugin, **kwargs):
-    print("label={}".format(payment['label']))
-    print("msat={}".format(payment['msat']))
-    print("preimage={}".format(payment['preimage']))
-
-    if payment['preimage'].endswith('0'):
-        # WIRE_TEMPORARY_NODE_FAILURE = 0x2002
-        return {'failure_message': "2002"}
-
-    return {'result': 'continue'}
-
-
-plugin.run()
diff --git a/tests/plugins/sendpay_notifications.py b/tests/plugins/sendpay_notifications.py
deleted file mode 100755
index 0f83290d..00000000
--- a/tests/plugins/sendpay_notifications.py
+++ /dev/null
@@ -1,34 +0,0 @@
-#!/usr/bin/env python3
-"""This plugin is used to check that sendpay_success and sendpay_failure calls are working correctly.
-"""
-from pyln.client import Plugin
-
-plugin = Plugin()
-
-
-@plugin.init()
-def init(configuration, options, plugin):
-    plugin.success_list = []
-    plugin.failure_list = []
-
-
-@plugin.subscribe("sendpay_success")
-def notify_sendpay_success(plugin, sendpay_success):
-    plugin.log("Received a sendpay_success: id={}, payment_hash={}".format(sendpay_success['id'], sendpay_success['payment_hash']))
-    plugin.success_list.append(sendpay_success)
-
-
-@plugin.subscribe("sendpay_failure")
-def notify_sendpay_failure(plugin, sendpay_failure):
-    plugin.log("Received a sendpay_failure: id={}, payment_hash={}".format(sendpay_failure['data']['id'],
-               sendpay_failure['data']['payment_hash']))
-    plugin.failure_list.append(sendpay_failure)
-
-
-@plugin.method('listsendpays_plugin')
-def record_lookup(plugin):
-    return {'sendpay_success': plugin.success_list,
-            'sendpay_failure': plugin.failure_list}
-
-
-plugin.run()
diff --git a/tests/plugins/shortcircuit.py b/tests/plugins/shortcircuit.py
deleted file mode 100755
index bdb088c1..00000000
--- a/tests/plugins/shortcircuit.py
+++ /dev/null
@@ -1,13 +0,0 @@
-#!/usr/bin/env python3
-
-from pyln.client import Plugin
-
-plugin = Plugin()
-
-
-@plugin.hook("htlc_accepted")
-def on_htlc_accepted(onion, htlc, plugin, **kwargs):
-    return {"result": "resolve", "payment_key": "00" * 32}
-
-
-plugin.run()
diff --git a/tests/plugins/utf8.py b/tests/plugins/utf8.py
deleted file mode 100755
index 16c3afe4..00000000
--- a/tests/plugins/utf8.py
+++ /dev/null
@@ -1,14 +0,0 @@
-#!/usr/bin/env python3
-from pyln.client import Plugin
-
-
-plugin = Plugin()
-
-
-@plugin.method("utf8")
-def echo(plugin, utf8):
-    assert '\\u' not in utf8
-    return {'utf8': utf8}
-
-
-plugin.run()
diff --git a/tests/plugins/validatejson.py b/tests/plugins/validatejson.py
deleted file mode 100755
index 8c32d4da..00000000
--- a/tests/plugins/validatejson.py
+++ /dev/null
@@ -1,12 +0,0 @@
-#!/usr/bin/env python3
-from pyln.client import Plugin
-
-plugin = Plugin()
-
-
-@plugin.method('validate-json-rpc')
-def validate_json_rpc(plugin, *args, **kwargs):
-    return {}
-
-
-plugin.run()
diff --git a/tests/test_connection.py b/tests/test_connection.py
index 19135be7..548c24ee 100644
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -4597,16 +4597,24 @@ def test_connect_ratelimit(node_factory, bitcoind):
 
 def test_onionmessage_forward_fail(node_factory, bitcoind):
     # The plugin will try to connect to l3, so it needs an advertized address.
-    l1, l2, l3 = node_factory.line_graph(3, wait_for_announce=True,
-                                         opts=[{},
-                                               {'dev-allow-localhost': None,
-                                                'may_reconnect': True,
-                                                'dev-no-reconnect': None,
-                                                'plugin': os.path.join(os.getcwd(), 'tests/plugins/onionmessage_forward_fail_notification.py'),
-                                                },
-                                               {'dev-allow-localhost': None,
-                                                'dev-no-reconnect': None,
-                                                'may_reconnect': True}])
+    def setup(plugin):
+        @plugin.subscribe("onionmessage_forward_fail")
+        def on_onionmessage_forward_fail(onionmessage_forward_fail, **kwargs):
+            plugin.log(f"Received onionmessage_forward_fail {onionmessage_forward_fail}")
+            plugin.rpc.connect(onionmessage_forward_fail['next_node_id'])
+            # injectonionmessage expects to unwrap, so hand it *incoming*
+            plugin.rpc.injectonionmessage(onionmessage_forward_fail['path_key'],
+                                          onionmessage_forward_fail['incoming'])
+
+    l1 = node_factory.get_node()
+    l2 = node_factory.get_node(inline_plugin=setup,
+                               options={'dev-allow-localhost': None,
+                                        'dev-no-reconnect': None},
+                               may_reconnect=True)
+    l3 = node_factory.get_node(options={'dev-allow-localhost': None,
+                                        'dev-no-reconnect': None},
+                               may_reconnect=True)
+    node_factory.join_nodes([l1, l2, l3], wait_for_announce=True)
 
     offer = l3.rpc.offer(300, "test_onionmessage_forward_fail")
     l2.rpc.disconnect(l3.info['id'], force=True)
@@ -4614,7 +4622,7 @@ def test_onionmessage_forward_fail(node_factory, bitcoind):
     # The plugin in l2 fixes up the connection, so this works!
     l1.rpc.fetchinvoice(offer['bolt12'])
 
-    l2.daemon.is_in_log('plugin-onionmessage_forward_fail_notification.py: Received onionmessage_forward_fail')
+    l2.daemon.is_in_log('plugin-inline-plugin.py: Received onionmessage_forward_fail')
 
 
 def test_private_channel_no_reconnect(node_factory):
diff --git a/tests/test_misc.py b/tests/test_misc.py
index 4154ad6e..560b53fb 100644
--- a/tests/test_misc.py
+++ b/tests/test_misc.py
@@ -999,12 +999,12 @@ def test_malformed_rpc(node_factory):
 def test_valid_json_cli(node_factory):
     """Make sure lightning-cli passes valid json values, so that rust and python plugins
     don't crash."""
-    l1 = node_factory.get_node(
-        options={
-            "log-level": "io",
-            "plugin": os.path.join(os.getcwd(), "tests/plugins/validatejson.py"),
-        }
-    )
+    def setup(plugin):
+        @plugin.method('validate-json-rpc')
+        def validate_json_rpc(plugin, *args, **kwargs):
+            return {}
+
+    l1 = node_factory.get_node(options={"log-level": "io"}, inline_plugin=setup)
     # If passed as a literal number rust's serde_json::from_str will fail as the
     # leading zero makes it invalid for an integer.
     nodeid = "030000000000000000000000000000000000000000000000000000000000000001"
@@ -1203,7 +1203,18 @@ def test_cli(node_factory):
 
 
 def test_cli_multiline_help(node_factory):
-    l1 = node_factory.get_node(options={'plugin': os.path.join(os.getcwd(), 'tests/plugins/multiline-help.py')})
+    def setup(plugin):
+        from pyln.client import Millisatoshi
+
+        @plugin.method("helpme")
+        def helpme(plugin, msat: Millisatoshi):
+            """This is a message which consumes multiple lines and thus should
+            be well-formatted by lightning-cli help
+
+            """
+            return {'help': msat}
+
+    l1 = node_factory.get_node(inline_plugin=setup)
 
     out = subprocess.check_output(['cli/lightning-cli',
                                    '--network={}'.format(TEST_NETWORK),
@@ -1431,15 +1442,40 @@ def test_funding_reorg_private(node_factory, bitcoind):
     """
     # Rescan to detect reorg at restart and may_reconnect so channeld
     # will restart.  Reorg can cause bad gossip msg.
-    opts = {'funding-confirms': 2, 'rescan': 10, 'may_reconnect': True,
-            'allow_bad_gossip': True,
-            # gossipd send lightning update for original channel.
-            'allow_warning': True,
-            'dev-fast-reconnect': None,
-            # if it's not zeroconf, we'll terminate on reorg.
-            'plugin': os.path.join(os.getcwd(), 'tests/plugins/zeroconf-selective.py'),
-            'zeroconf_allow': 'any'}
-    l1, l2 = node_factory.line_graph(2, fundchannel=False, opts=opts)
+    def setup(plugin):
+        plugin.add_option(
+            'zeroconf_allow',
+            '03864ef025fde8fb587d989186ce6a4a186895ee44a926bfc370e2c366597a3f8f',
+            'A node_id to allow zeroconf channels from',
+        )
+        plugin.add_option(
+            'zeroconf_mindepth',
+            0,
+            'Number of confirmations to require from allowlisted peers',
+        )
+
+        @plugin.hook('openchannel')
+        def on_openchannel(openchannel, plugin, **kwargs):
+            plugin.log(repr(openchannel))
+            mindepth = int(plugin.options['zeroconf_mindepth']['value'])
+
+            if openchannel['id'] == plugin.options['zeroconf_allow']['value'] or plugin.options['zeroconf_allow']['value'] == 'any':
+                plugin.log(f"This peer is in the zeroconf allowlist, setting mindepth={mindepth}")
+                return {'result': 'continue', 'mindepth': mindepth}
+            else:
+                return {'result': 'continue'}
+
+    cli_opts = {'funding-confirms': 2, 'rescan': 10,
+                'dev-fast-reconnect': None,
+                # if it's not zeroconf, we'll terminate on reorg.
+                'zeroconf_allow': 'any'}
+    node_opts = {'may_reconnect': True,
+                 'allow_bad_gossip': True,
+                 # gossipd send lightning update for original channel.
+                 'allow_warning': True}
+    l1 = node_factory.get_node(inline_plugin=setup, options=cli_opts, **node_opts)
+    l2 = node_factory.get_node(inline_plugin=setup, options=cli_opts, **node_opts)
+    node_factory.join_nodes([l1, l2], fundchannel=False)
     l1.fundwallet(10000000)
     sync_blockheight(bitcoind, [l1])                # height 102
     bitcoind.generate_block(3)                      # heights 103-105
@@ -1477,12 +1513,36 @@ def test_funding_reorg_remote_lags(node_factory, bitcoind):
     """Nodes may disagree about short_channel_id before channel announcement
     """
     # may_reconnect so channeld will restart; bad gossip can happen due to reorg
-    opts = {'funding-confirms': 1, 'may_reconnect': True, 'allow_bad_gossip': True,
-            'allow_warning': True, 'dev-fast-reconnect': None,
-            # if it's not zeroconf, l2 will terminate on reorg.
-            'plugin': os.path.join(os.getcwd(), 'tests/plugins/zeroconf-selective.py'),
-            'zeroconf_allow': 'any'}
-    l1, l2 = node_factory.line_graph(2, fundchannel=False, opts=opts)
+    def setup(plugin):
+        plugin.add_option(
+            'zeroconf_allow',
+            '03864ef025fde8fb587d989186ce6a4a186895ee44a926bfc370e2c366597a3f8f',
+            'A node_id to allow zeroconf channels from',
+        )
+        plugin.add_option(
+            'zeroconf_mindepth',
+            0,
+            'Number of confirmations to require from allowlisted peers',
+        )
+
+        @plugin.hook('openchannel')
+        def on_openchannel(openchannel, plugin, **kwargs):
+            plugin.log(repr(openchannel))
+            mindepth = int(plugin.options['zeroconf_mindepth']['value'])
+
+            if openchannel['id'] == plugin.options['zeroconf_allow']['value'] or plugin.options['zeroconf_allow']['value'] == 'any':
+                plugin.log(f"This peer is in the zeroconf allowlist, setting mindepth={mindepth}")
+                return {'result': 'continue', 'mindepth': mindepth}
+            else:
+                return {'result': 'continue'}
+
+    cli_opts = {'funding-confirms': 1, 'dev-fast-reconnect': None,
+                # if it's not zeroconf, l2 will terminate on reorg.
+                'zeroconf_allow': 'any'}
+    node_opts = {'may_reconnect': True, 'allow_bad_gossip': True, 'allow_warning': True}
+    l1 = node_factory.get_node(inline_plugin=setup, options=cli_opts, **node_opts)
+    l2 = node_factory.get_node(inline_plugin=setup, options=cli_opts, **node_opts)
+    node_factory.join_nodes([l1, l2], fundchannel=False)
     l1.fundwallet(10000000)
     sync_blockheight(bitcoind, [l1])                # height 102
 
diff --git a/tests/test_pay.py b/tests/test_pay.py
index c68b4ba3..e0b68f09 100644
--- a/tests/test_pay.py
+++ b/tests/test_pay.py
@@ -3020,7 +3020,16 @@ def test_sendonion_rpc(node_factory):
 @pytest.mark.openchannel('v2')
 def test_partial_payment(node_factory, bitcoind, executor):
     # We want to test two payments at the same time, before we send commit
-    l1, l2, l3, l4 = node_factory.get_nodes(4, [{}] + [{'dev-disable-commit-after': 0, 'dev-no-htlc-timeout': None}] * 2 + [{'plugin': os.path.join(os.getcwd(), 'tests/plugins/print_htlc_onion.py')}])
+    def setup(plugin):
+        @plugin.hook("htlc_accepted")
+        def on_htlc_accepted(htlc, onion, plugin, **kwargs):
+            plugin.log("Got onion {}".format(onion))
+            return {'result': 'continue'}
+
+    l1 = node_factory.get_node()
+    l2 = node_factory.get_node(options={'dev-disable-commit-after': 0, 'dev-no-htlc-timeout': None})
+    l3 = node_factory.get_node(options={'dev-disable-commit-after': 0, 'dev-no-htlc-timeout': None})
+    l4 = node_factory.get_node(inline_plugin=setup)
 
     # Two routes to l4: one via l2, and one via l3.
     l1.rpc.connect(l2.info['id'], 'localhost', l2.port)
@@ -3156,7 +3165,7 @@ def test_partial_payment(node_factory, bitcoind, executor):
     assert res['partid'] == 2
 
     for i in range(2):
-        line = l4.daemon.wait_for_log('print_htlc_onion.py: Got onion')
+        line = l4.daemon.wait_for_log('inline-plugin.py: Got onion')
         assert "'type': 'tlv'" in line
         assert "'forward_msat': 499" in line or "'forward_msat': 501" in line
         assert "'total_msat': 1000" in line
@@ -3762,12 +3771,22 @@ def test_invalid_onion_channel_update(node_factory):
     even if some remote node does not send the required
     `channel_update`.
     '''
-    plugin = os.path.join(os.getcwd(), 'tests/plugins/fail_htlcs_invalid.py')
-    l1, l2, l3 = node_factory.line_graph(3,
-                                         opts=[{},
-                                               {'plugin': plugin},
-                                               {}],
-                                         wait_for_announce=True)
+    def setup(plugin):
+        @plugin.hook("htlc_accepted")
+        def on_htlc_accepted(onion, plugin, **kwargs):
+            plugin.log("Failing htlc on purpose with invalid onion failure")
+            plugin.log("onion: %r" % (onion))
+            # WIRE_TEMPORARY_CHANNEL_FAILURE = 0x1007
+            # This failure code should be followed by a
+            # `channel_update`; we deliberately return
+            # a 0-length `channel_update` to trigger
+            # issue #3757 reported by @sumBTC.
+            return {"result": "fail", "failure_message": "10070000"}
+
+    l1 = node_factory.get_node()
+    l2 = node_factory.get_node(inline_plugin=setup)
+    l3 = node_factory.get_node()
+    node_factory.join_nodes([l1, l2, l3], wait_for_announce=True)
 
     l1id = l1.info['id']
 
@@ -5893,11 +5912,26 @@ def test_blinded_reply_path_scid(node_factory):
 
 
 def test_pay_while_opening_channel(node_factory, bitcoind, executor):
-    delay_plugin = {'plugin': os.path.join(os.getcwd(),
-                                           'tests/plugins/openchannel_hook_delay.py'),
-                    'delaytime': '10'}
+    def setup(plugin):
+        import time
+        plugin.add_option('delaytime', '10', 'How long to hold the WIRE_OPEN_CHANNEL.')
+
+        @plugin.hook('openchannel')
+        def on_openchannel(openchannel, plugin, **kwargs):
+            delaytime = float(plugin.get_option('delaytime'))
+            plugin.log(f'delaying WIRE_ACCEPT_CHANNEL for {delaytime}s')
+            time.sleep(delaytime)
+            return {'result': 'continue'}
+
+        @plugin.hook('openchannel2')
+        def on_openchannel2(openchannel2, plugin, **kwargs):
+            delaytime = float(plugin.get_option('delaytime'))
+            plugin.log(f'delaying WIRE_ACCEPT_CHANNEL for {delaytime}s')
+            time.sleep(delaytime)
+            return {'result': 'continue'}
+
     l1, l2 = node_factory.line_graph(2, fundamount=10**6, wait_for_announce=True)
-    l3 = node_factory.get_node(options=delay_plugin)
+    l3 = node_factory.get_node(inline_plugin=setup, options={'delaytime': '10'})
     l1.connect(l3)
     executor.submit(l1.rpc.fundchannel, l3.info['id'], 100000)
     wait_for(lambda: l1.rpc.listpeerchannels(l3.info['id'])['channels'] != [])
diff --git a/tests/test_plugin.py b/tests/test_plugin.py
index 69c1302b..52806355 100644
--- a/tests/test_plugin.py
+++ b/tests/test_plugin.py
@@ -667,8 +667,13 @@ def test_db_hook_multiple(node_factory, executor):
 
 
 def test_utf8_passthrough(node_factory, executor):
-    l1 = node_factory.get_node(options={'plugin': os.path.join(os.getcwd(), 'tests/plugins/utf8.py'),
-                                        'log-level': 'io'})
+    def setup(plugin):
+        @plugin.method("utf8")
+        def echo(plugin, utf8):
+            assert '\\u' not in utf8
+            return {'utf8': utf8}
+
+    l1 = node_factory.get_node(inline_plugin=setup, options={'log-level': 'io'})
 
     # This works because Python unmangles.
     res = l1.rpc.call('utf8', ['ナンセンス 1杯'])
@@ -688,8 +693,22 @@ def test_utf8_passthrough(node_factory, executor):
 def test_invoice_payment_hook(node_factory):
     """ l1 uses the reject-payment plugin to reject invoices with odd preimages.
     """
-    opts = [{}, {'plugin': os.path.join(os.getcwd(), 'tests/plugins/reject_some_invoices.py')}]
-    l1, l2 = node_factory.line_graph(2, opts=opts)
+    def setup(plugin):
+        @plugin.hook('invoice_payment')
+        def on_payment(payment, plugin, **kwargs):
+            plugin.log("label={}".format(payment['label']))
+            plugin.log("msat={}".format(payment['msat']))
+            plugin.log("preimage={}".format(payment['preimage']))
+
+            if payment['preimage'].endswith('0'):
+                # WIRE_TEMPORARY_NODE_FAILURE = 0x2002
+                return {'failure_message': "2002"}
+
+            return {'result': 'continue'}
+
+    l1 = node_factory.get_node()
+    l2 = node_factory.get_node(inline_plugin=setup)
+    node_factory.join_nodes([l1, l2])
 
     # This one works
     inv1 = l2.rpc.invoice(1230, 'label', 'description', preimage='1' * 64)
@@ -731,8 +750,31 @@ def test_invoice_payment_hook_hold(node_factory, executor):
 def test_openchannel_hook(node_factory, bitcoind):
     """ l2 uses the reject_odd_funding_amounts plugin to reject some openings.
     """
-    opts = [{}, {'plugin': os.path.join(os.getcwd(), 'tests/plugins/reject_odd_funding_amounts.py')}]
-    l1, l2 = node_factory.line_graph(2, fundchannel=False, opts=opts)
+    def setup(plugin):
+        from pyln.client import Millisatoshi
+
+        def run_check(funding_amt_str):
+            if Millisatoshi(funding_amt_str).to_satoshi() % 2 == 1:
+                return {'result': 'reject', 'error_message': "I don't like odd amounts"}
+            return {'result': 'continue'}
+
+        @plugin.hook('openchannel')
+        def on_openchannel(openchannel, plugin, **kwargs):
+            plugin.log("{} VARS".format(len(openchannel.keys())))
+            for k in sorted(openchannel.keys()):
+                plugin.log("{}={}".format(k, openchannel[k]))
+            return run_check(openchannel['funding_msat'])
+
+        @plugin.hook('openchannel2')
+        def on_openchannel2(openchannel2, plugin, **kwargs):
+            plugin.log("{} VARS".format(len(openchannel2.keys())))
+            for k in sorted(openchannel2.keys()):
+                plugin.log("{}={}".format(k, openchannel2[k]))
+            return run_check(openchannel2['their_funding_msat'])
+
+    l1 = node_factory.get_node()
+    l2 = node_factory.get_node(inline_plugin=setup)
+    node_factory.join_nodes([l1, l2], fundchannel=False)
     l1.fundwallet(10**6)
 
     # Even amount: works.
@@ -776,9 +818,9 @@ def test_openchannel_hook(node_factory, bitcoind):
             'push_msat': 0,
         })
 
-    l2.daemon.wait_for_log('reject_odd_funding_amounts.py: {} VARS'.format(len(expected)))
+    l2.daemon.wait_for_log('inline-plugin.py: {} VARS'.format(len(expected)))
     for k, v in expected.items():
-        assert l2.daemon.is_in_log('reject_odd_funding_amounts.py: {}={}'.format(k, v))
+        assert l2.daemon.is_in_log('inline-plugin.py: {}={}'.format(k, v))
 
     # Close it.
     txid = only_one(l1.rpc.close(l2.info['id'])['txids'])
@@ -1252,11 +1294,15 @@ def test_htlc_accepted_hook_fail(node_factory):
 def test_htlc_accepted_hook_resolve(node_factory):
     """l3 creates an invoice, l2 knows the preimage and will shortcircuit.
     """
-    l1, l2, l3 = node_factory.line_graph(3, opts=[
-        {},
-        {'plugin': os.path.join(os.getcwd(), 'tests/plugins/shortcircuit.py')},
-        {}
-    ], wait_for_announce=True)
+    def setup(plugin):
+        @plugin.hook("htlc_accepted")
+        def on_htlc_accepted(onion, htlc, plugin, **kwargs):
+            return {"result": "resolve", "payment_key": "00" * 32}
+
+    l1 = node_factory.get_node()
+    l2 = node_factory.get_node(inline_plugin=setup)
+    l3 = node_factory.get_node()
+    node_factory.join_nodes([l1, l2, l3], wait_for_announce=True)
 
     inv = l3.rpc.invoice(amount_msat=1000, label="lbl", description="desc", preimage="00" * 32)['bolt11']
     l1.rpc.xpay(inv)
@@ -1408,41 +1454,62 @@ def test_htlc_accepted_hook_forward_restart(node_factory, executor):
 def test_warning_notification(node_factory):
     """ test 'warning' notifications
     """
-    l1 = node_factory.get_node(options={'plugin': os.path.join(os.getcwd(), 'tests/plugins/pretend_badlog.py')}, broken_log=r'Test warning notification\(for broken event\)|LINE[12]')
+    def setup(plugin):
+        @plugin.init()
+        def init(configuration, options, plugin):
+            plugin.log("initialized")
+
+        @plugin.subscribe("warning")
+        def notify_warning(plugin, warning, **kwargs):
+            plugin.log("Received warning")
+            plugin.log("level: {}".format(warning['level']))
+            plugin.log("time: {}".format(warning['time']))
+            plugin.log("source: {}".format(warning['source']))
+            plugin.log("log: {}".format(warning['log']))
+
+        @plugin.method("pretendbad")
+        def pretend_bad(event, level, plugin):
+            """Log an specified level entry.
+            And in plugin, we use 'warn'/'error' instead of
+            'unusual'/'broken'
+            """
+            plugin.log("{}".format(event), level)
+
+    l1 = node_factory.get_node(inline_plugin=setup, broken_log=r'Test warning notification\(for broken event\)|LINE[12]')
 
     # 1. test 'warn' level
     event = "Test warning notification(for unusual event)"
     l1.rpc.call('pretendbad', {'event': event, 'level': 'warn'})
 
     # ensure an unusual log_entry was produced by 'pretendunusual' method
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: Test warning notification\\(for unusual event\\)')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: Test warning notification\\(for unusual event\\)')
 
     # now wait for notification
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: Received warning')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: level: warn')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: time: *')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: source: plugin-pretend_badlog.py')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: log: Test warning notification\\(for unusual event\\)')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: Received warning')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: level: warn')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: time: *')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: source: plugin-inline-plugin.py')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: log: Test warning notification\\(for unusual event\\)')
 
     # 2. test 'error' level, steps like above
     event = "Test warning notification(for broken event)"
     l1.rpc.call('pretendbad', {'event': event, 'level': 'error'})
-    l1.daemon.wait_for_log(r'\*\*BROKEN\*\* plugin-pretend_badlog.py: Test warning notification\(for broken event\)')
+    l1.daemon.wait_for_log(r'\*\*BROKEN\*\* plugin-inline-plugin.py: Test warning notification\(for broken event\)')
 
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: Received warning')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: level: error')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: time: *')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: source: plugin-pretend_badlog.py')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: log: Test warning notification\\(for broken event\\)')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: Received warning')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: level: error')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: time: *')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: source: plugin-inline-plugin.py')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: log: Test warning notification\\(for broken event\\)')
 
     # Test linesplitting while we're here
     l1.rpc.call('pretendbad', {'event': 'LINE1\nLINE2', 'level': 'error'})
-    l1.daemon.wait_for_log(r'\*\*BROKEN\*\* plugin-pretend_badlog.py: LINE1')
-    l1.daemon.wait_for_log(r'\*\*BROKEN\*\* plugin-pretend_badlog.py: LINE2')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: Received warning')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: log: LINE1')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: Received warning')
-    l1.daemon.wait_for_log('plugin-pretend_badlog.py: log: LINE2')
+    l1.daemon.wait_for_log(r'\*\*BROKEN\*\* plugin-inline-plugin.py: LINE1')
+    l1.daemon.wait_for_log(r'\*\*BROKEN\*\* plugin-inline-plugin.py: LINE2')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: Received warning')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: log: LINE1')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: Received warning')
+    l1.daemon.wait_for_log('plugin-inline-plugin.py: log: LINE2')
 
 
 def test_invoice_payment_notification(node_factory):
@@ -1632,11 +1699,33 @@ def test_forward_event_notification(node_factory, bitcoind, executor):
 def test_sendpay_notifications(node_factory, bitcoind):
     """ test 'sendpay_success' and 'sendpay_failure' notifications
     """
+    def setup(plugin):
+        @plugin.init()
+        def init(configuration, options, plugin):
+            plugin.success_list = []
+            plugin.failure_list = []
+
+        @plugin.subscribe("sendpay_success")
+        def notify_sendpay_success(plugin, sendpay_success):
+            plugin.log("Received a sendpay_success: id={}, payment_hash={}".format(sendpay_success['id'], sendpay_success['payment_hash']))
+            plugin.success_list.append(sendpay_success)
+
+        @plugin.subscribe("sendpay_failure")
+        def notify_sendpay_failure(plugin, sendpay_failure):
+            plugin.log("Received a sendpay_failure: id={}, payment_hash={}".format(sendpay_failure['data']['id'],
+                       sendpay_failure['data']['payment_hash']))
+            plugin.failure_list.append(sendpay_failure)
+
+        @plugin.method('listsendpays_plugin')
+        def record_lookup(plugin):
+            return {'sendpay_success': plugin.success_list,
+                    'sendpay_failure': plugin.failure_list}
+
     amount = 10**8
-    opts = [{'plugin': os.path.join(os.getcwd(), 'tests/plugins/sendpay_notifications.py')},
-            {},
-            {'may_reconnect': False}]
-    l1, l2, l3 = node_factory.line_graph(3, opts=opts, wait_for_announce=True)
+    l1 = node_factory.get_node(inline_plugin=setup)
+    l2 = node_factory.get_node()
+    l3 = node_factory.get_node(may_reconnect=False)
+    node_factory.join_nodes([l1, l2, l3], wait_for_announce=True)
     chanid23 = l2.get_channel_scid(l3)
 
     inv1 = l3.rpc.invoice(amount, "first", "desc")
@@ -1663,10 +1752,32 @@ def test_sendpay_notifications(node_factory, bitcoind):
 
 
 def test_sendpay_notifications_nowaiter(node_factory):
-    opts = [{'plugin': os.path.join(os.getcwd(), 'tests/plugins/sendpay_notifications.py')},
-            {},
-            {'may_reconnect': False}]
-    l1, l2, l3 = node_factory.line_graph(3, opts=opts, wait_for_announce=True)
+    def setup(plugin):
+        @plugin.init()
+        def init(configuration, options, plugin):
+            plugin.success_list = []
+            plugin.failure_list = []
+
+        @plugin.subscribe("sendpay_success")
+        def notify_sendpay_success(plugin, sendpay_success):
+            plugin.log("Received a sendpay_success: id={}, payment_hash={}".format(sendpay_success['id'], sendpay_success['payment_hash']))
+            plugin.success_list.append(sendpay_success)
+
+        @plugin.subscribe("sendpay_failure")
+        def notify_sendpay_failure(plugin, sendpay_failure):
+            plugin.log("Received a sendpay_failure: id={}, payment_hash={}".format(sendpay_failure['data']['id'],
+                       sendpay_failure['data']['payment_hash']))
+            plugin.failure_list.append(sendpay_failure)
+
+        @plugin.method('listsendpays_plugin')
+        def record_lookup(plugin):
+            return {'sendpay_success': plugin.success_list,
+                    'sendpay_failure': plugin.failure_list}
+
+    l1 = node_factory.get_node(inline_plugin=setup)
+    l2 = node_factory.get_node()
+    l3 = node_factory.get_node(may_reconnect=False)
+    node_factory.join_nodes([l1, l2, l3], wait_for_announce=True)
     chanid23 = l2.get_channel_scid(l3)
     amount = 10**8
 
@@ -3428,10 +3539,26 @@ def test_autoclean_once(node_factory):
 def test_block_added_notifications(node_factory, bitcoind):
     """Test if a plugin gets notifications when a new block is found"""
     base = bitcoind.rpc.getblockchaininfo()["blocks"]
-    plugin = [
-        os.path.join(os.getcwd(), "tests/plugins/block_added.py"),
-    ]
-    l1 = node_factory.get_node(options={"plugin": plugin})
+
+    def make_setup():
+        blocks_catched = []
+
+        def setup(plugin):
+            @plugin.init()
+            def on_init(plugin, options, configuration, **kwargs):
+                blocks_catched.clear()
+
+            @plugin.subscribe("block_added")
+            def notify_block_added(plugin, block_added, **kwargs):
+                blocks_catched.append(block_added["height"])
+
+            @plugin.method("blockscatched")
+            def return_moves(plugin):
+                return blocks_catched
+
+        return setup
+
+    l1 = node_factory.get_node(inline_plugin=make_setup())
     ret = l1.rpc.call("blockscatched")
     assert len(ret) == 1 and ret[0] == base + 0
 
@@ -3440,7 +3567,7 @@ def test_block_added_notifications(node_factory, bitcoind):
     ret = l1.rpc.call("blockscatched")
     assert len(ret) == 3 and ret[0] == base + 0 and ret[2] == base + 2
 
-    l2 = node_factory.get_node(options={"plugin": plugin})
+    l2 = node_factory.get_node(inline_plugin=make_setup())
     ret = l2.rpc.call("blockscatched")
     assert len(ret) == 1 and ret[0] == base + 2
 
diff --git a/tests/test_xpay.py b/tests/test_xpay.py
index c6f8af0e..4c4cec3f 100644
--- a/tests/test_xpay.py
+++ b/tests/test_xpay.py
@@ -7,7 +7,6 @@ from utils import (
     sync_blockheight,
 )
 
-import ast
 import os
 import pytest
 import re
@@ -826,9 +825,46 @@ def test_attempt_notifications(node_factory):
         # other types are ignored
         return obj
 
-    plugin_path = os.path.join(os.getcwd(), 'tests/plugins/custom_notifications.py')
-    l1, l2, l3 = node_factory.line_graph(3, wait_for_announce=True,
-                                         opts=[{"plugin": plugin_path}, {}, {}])
+    part_starts = []
+    part_ends = []
+
+    def setup(plugin):
+        @plugin.subscribe("pay_part_start")
+        def on_pay_part_start(origin, **kwargs):
+            part_starts.append(kwargs)
+
+        @plugin.subscribe("pay_part_end")
+        def on_pay_part_end(origin, **kwargs):
+            part_ends.append(kwargs)
+
+        @plugin.subscribe("pay_success")
+        def on_pay_success(origin, pay_success, **kwargs):
+            pass
+
+        @plugin.subscribe("custom")
+        def on_custom_notification(origin, message, **kwargs):
+            pass
+
+        @plugin.subscribe("ididntannouncethis")
+        def on_faulty_emit(origin, payload, **kwargs):
+            pass
+
+        @plugin.method("emit")
+        def emit(plugin):
+            """Emit a simple string notification to topic "custom" """
+            plugin.notify("custom", {'message': "Hello world"})
+
+        @plugin.method("faulty-emit")
+        def faulty_emit(plugin):
+            """Emit a simple string notification to topic "custom" """
+            plugin.notify("ididntannouncethis", {'message': "Hello world"})
+
+        plugin.add_notification_topic("custom")
+
+    l1 = node_factory.get_node(inline_plugin=setup)
+    l2 = node_factory.get_node()
+    l3 = node_factory.get_node()
+    node_factory.join_nodes([l1, l2, l3], wait_for_announce=True)
 
     scid12 = only_one(l1.rpc.listpeerchannels(l2.info['id'])['channels'])['short_channel_id']
     scid12_dir = only_one(l1.rpc.listpeerchannels(l2.info['id'])['channels'])['direction']
@@ -837,9 +873,8 @@ def test_attempt_notifications(node_factory):
     inv1 = l3.rpc.invoice(5000000, 'test_attempt_notifications1', 'test_attempt_notifications1')
     l1.rpc.xpay(inv1['bolt11'])
 
-    line = l1.daemon.wait_for_log("plugin-custom_notifications.py: Got pay_part_start: ")
-    dict_str = line.split("Got pay_part_start: ", 1)[1]
-    data = zero_fields(ast.literal_eval(dict_str), ['groupid'])
+    wait_for(lambda: len(part_starts) >= 1)
+    data = zero_fields(part_starts.pop(0), ['groupid'])
     expected = {'pay_part_start':
                 {'payment_hash': inv1['payment_hash'],
                  'groupid': 0,
@@ -858,9 +893,8 @@ def test_attempt_notifications(node_factory):
                            'channel_out_msat': 5000000}]}}
     assert data == expected
 
-    line = l1.daemon.wait_for_log("plugin-custom_notifications.py: Got pay_part_end: ")
-    dict_str = line.split("Got pay_part_end: ", 1)[1]
-    data = zero_fields(ast.literal_eval(dict_str), ('duration', 'groupid'))
+    wait_for(lambda: len(part_ends) >= 1)
+    data = zero_fields(part_ends.pop(0), ('duration', 'groupid'))
     expected = {'pay_part_end':
                 {'payment_hash': inv1['payment_hash'],
                  'status': 'success',
@@ -876,9 +910,8 @@ def test_attempt_notifications(node_factory):
     with pytest.raises(RpcError, match=r"Destination said it doesn't know invoice: incorrect_or_unknown_payment_details"):
         l1.rpc.xpay(inv2['bolt11'])
 
-    line = l1.daemon.wait_for_log("plugin-custom_notifications.py: Got pay_part_start: ")
-    dict_str = line.split("Got pay_part_start: ", 1)[1]
-    data = zero_fields(ast.literal_eval(dict_str), ['groupid'])
+    wait_for(lambda: len(part_starts) >= 1)
+    data = zero_fields(part_starts.pop(0), ['groupid'])
     expected = {'pay_part_start':
                 {'payment_hash': inv2['payment_hash'],
                  'groupid': 0,
@@ -897,9 +930,8 @@ def test_attempt_notifications(node_factory):
                            'channel_out_msat': 10000000}]}}
     assert data == expected
 
-    line = l1.daemon.wait_for_log("plugin-custom_notifications.py: Got pay_part_end: ")
-    dict_str = line.split("Got pay_part_end: ", 1)[1]
-    data = zero_fields(ast.literal_eval(dict_str), ('duration', 'groupid'))
+    wait_for(lambda: len(part_ends) >= 1)
+    data = zero_fields(part_ends.pop(0), ('duration', 'groupid'))
     expected = {'pay_part_end':
                 {'payment_hash': inv2['payment_hash'],
                  'status': 'failure',
@@ -917,9 +949,8 @@ def test_attempt_notifications(node_factory):
     with pytest.raises(RpcError, match=r"Failed after 1 attempts"):
         l1.rpc.xpay(inv2['bolt11'])
 
-    line = l1.daemon.wait_for_log("plugin-custom_notifications.py: Got pay_part_start: ")
-    dict_str = line.split("Got pay_part_start: ", 1)[1]
-    data = zero_fields(ast.literal_eval(dict_str), ['groupid'])
+    wait_for(lambda: len(part_starts) >= 1)
+    data = zero_fields(part_starts.pop(0), ['groupid'])
     expected = {'pay_part_start':
                 {'payment_hash': inv2['payment_hash'],
                  'groupid': 0,
@@ -938,9 +969,8 @@ def test_attempt_notifications(node_factory):
                            'channel_out_msat': 10000000}]}}
     assert data == expected
 
-    line = l1.daemon.wait_for_log("plugin-custom_notifications.py: Got pay_part_end: ")
-    dict_str = line.split("Got pay_part_end: ", 1)[1]
-    data = zero_fields(ast.literal_eval(dict_str), ('duration', 'groupid', 'failed_msg'))
+    wait_for(lambda: len(part_ends) >= 1)
+    data = zero_fields(part_ends.pop(0), ('duration', 'groupid', 'failed_msg'))
     expected = {'pay_part_end':
                 {'payment_hash': inv2['payment_hash'],
                  'status': 'failure',