pytest: test xpay notifications.

diff --git a/tests/plugins/custom_notifications.py b/tests/plugins/custom_notifications.py
index a3fa9b9a..8ece850d 100755
--- a/tests/plugins/custom_notifications.py
+++ b/tests/plugins/custom_notifications.py
@@ -34,6 +34,16 @@ def on_pay_success(origin, payload, **kwargs):
     )
 
 
+@plugin.subscribe("pay_part_start")
+def on_pay_part_start(origin, payload, **kwargs):
+    plugin.log("Got pay_part_start: {}".format(payload))
+
+
+@plugin.subscribe("pay_part_end")
+def on_pay_part_end(origin, payload, **kwargs):
+    plugin.log("Got pay_part_end: {}".format(payload))
+
+
 @plugin.subscribe("ididntannouncethis")
 def on_faulty_emit(origin, payload, **kwargs):
     """We should never receive this as it gets dropped.
diff --git a/tests/test_xpay.py b/tests/test_xpay.py
index f99764ea..9022e28d 100644
--- a/tests/test_xpay.py
+++ b/tests/test_xpay.py
@@ -832,3 +832,53 @@ lightning-cli pay lni1qqgv5nalmz08ukj4av074kyk6pepq93pqvvhnlnvurnfanndnxjtcjnmxr
     # This doesn't!
     l1.rpc.xpay(inv)
     l1.daemon.wait_for_log(f'Adding HTLC 1 amount=15002msat cltv={110 + 1 + 100 + 200 + 400}')
+
+
+def test_attempt_notifications(node_factory):
+    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}, {}, {}])
+
+    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']
+    scid23 = only_one(l2.rpc.listpeerchannels(l3.info['id'])['channels'])['short_channel_id']
+    scid23_dir = only_one(l2.rpc.listpeerchannels(l3.info['id'])['channels'])['direction']
+    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: ")
+    regex = r".*Got pay_part_start: \{'payment_hash': '" + inv1['payment_hash'] + r"', 'groupid': [0-9]*, 'partid': 1, 'total_payment_msat': 5000000, 'attempt_msat': 5000000, 'hops': \[\{'next_node': '" + l2.info['id'] + r"', 'short_channel_id': '" + scid12 + r"', 'direction': " + str(scid12_dir) + r", 'channel_in_msat': 5000051, 'channel_out_msat': 5000051\}, \{'next_node': '" + l3.info['id'] + r"', 'short_channel_id': '" + scid23 + r"', 'direction': " + str(scid23_dir) + r", 'channel_in_msat': 5000051, 'channel_out_msat': 5000000\}\]\}"
+    assert re.match(regex, line)
+
+    # Note, duration always has 9 decimals, EXCEPT that the python code interprets it, so if the last digit is a 0 it will only print 8.
+    line = l1.daemon.wait_for_log("plugin-custom_notifications.py: Got pay_part_end: ")
+    regex = r".*Got pay_part_end: \{'status': 'success', 'duration': [0-9]*\.[0-9]*, 'payment_hash': '" + inv1['payment_hash'] + r"', 'groupid': [0-9]*, 'partid': 1\}"
+    assert re.match(regex, line)
+
+    inv2 = l3.rpc.invoice(10000000, 'test_attempt_notifications2', 'test_attempt_notifications2')
+    l3.rpc.delinvoice('test_attempt_notifications2', "unpaid")
+
+    # Final node failure
+    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: ")
+    regex = r".*Got pay_part_start: \{'payment_hash': '" + inv2['payment_hash'] + r"', 'groupid': [0-9]*, 'partid': 1, 'total_payment_msat': 10000000, 'attempt_msat': 10000000, 'hops': \[\{'next_node': '" + l2.info['id'] + r"', 'short_channel_id': '" + scid12 + r"', 'direction': " + str(scid12_dir) + r", 'channel_in_msat': 10000101, 'channel_out_msat': 10000101\}, \{'next_node': '" + l3.info['id'] + r"', 'short_channel_id': '" + scid23 + r"', 'direction': " + str(scid23_dir) + r", 'channel_in_msat': 10000101, 'channel_out_msat': 10000000\}\]\}"
+    assert re.match(regex, line)
+
+    line = l1.daemon.wait_for_log("plugin-custom_notifications.py: Got pay_part_end: ")
+    regex = r".*Got pay_part_end: \{'status': 'failure', 'payment_hash': '" + inv2['payment_hash'] + r"', 'groupid': [0-9]*, 'partid': 1, 'failed_msg': '400f00000000009896800000006c', 'duration': [0-9]*\.[0-9]*, 'failed_node_id': '" + l3.info['id'] + r"', 'error_code': 16399, 'error_message': 'incorrect_or_unknown_payment_details'\}"
+    assert re.match(regex, line)
+
+    # Intermediary node failure
+    l3.stop()
+    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: ")
+    regex = r".*Got pay_part_start: \{'payment_hash': '" + inv2['payment_hash'] + r"', 'groupid': [0-9]*, 'partid': 1, 'total_payment_msat': 10000000, 'attempt_msat': 10000000, 'hops': \[\{'next_node': '" + l2.info['id'] + r"', 'short_channel_id': '" + scid12 + r"', 'direction': " + str(scid12_dir) + r", 'channel_in_msat': 10000101, 'channel_out_msat': 10000101\}, \{'next_node': '" + l3.info['id'] + r"', 'short_channel_id': '" + scid23 + r"', 'direction': " + str(scid23_dir) + r", 'channel_in_msat': 10000101, 'channel_out_msat': 10000000\}\]\}"
+    assert re.match(regex, line)
+
+    line = l1.daemon.wait_for_log("plugin-custom_notifications.py: Got pay_part_end: ")
+    regex = r".*Got pay_part_end: \{'status': 'failure', 'payment_hash': '" + inv2['payment_hash'] + r"', 'groupid': [0-9]*, 'partid': 1, 'failed_msg': '1007[a-f0-9]*', 'duration': [0-9]*\.[0-9]*, 'failed_node_id': '" + l2.info['id'] + r"', 'failed_short_channel_id': '" + scid23 + r"', 'failed_direction': " + str(scid23_dir) + r", 'error_code': 4103, 'error_message': 'temporary_channel_failure'\}"
+    assert re.match(regex, line)