More pytest optimizations

contrib/pyln-testing/pyln/testing/fixtures.py

@@ -174,6 +174,7 @@ def teardown_checks(request):
 @pytest.fixture
 def node_factory(request, directory, test_name, bitcoind, executor, db_provider, teardown_checks, node_cls):
     nf = NodeFactory(
+        request,
         test_name,
         bitcoind,
         executor,

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

@@ -550,9 +550,10 @@ def wait(self, timeout=10):
 
 
 class LightningNode(object):
-    def __init__(self, node_id, lightning_dir, bitcoind, executor, may_fail=False,
+    def __init__(self, node_id, lightning_dir, bitcoind, executor, valgrind, may_fail=False,
                  may_reconnect=False, allow_broken_log=False,
-                 allow_bad_gossip=False, db=None, port=None, disconnect=None, random_hsm=None, options=None, **kwargs):
+                 allow_bad_gossip=False, db=None, port=None, disconnect=None, random_hsm=None, options=None,
+                 **kwargs):
         self.bitcoin = bitcoind
         self.executor = executor
         self.may_fail = may_fail
@@ -585,7 +586,7 @@ def __init__(self, node_id, lightning_dir, bitcoind, executor, may_fail=False,
             self.daemon.env["LIGHTNINGD_DEV_MEMLEAK"] = "1"
             if os.getenv("DEBUG_SUBD"):
                 self.daemon.opts["dev-debugger"] = os.getenv("DEBUG_SUBD")
-            if VALGRIND:
+            if valgrind:
                 self.daemon.env["LIGHTNINGD_DEV_NO_BACKTRACE"] = "1"
             if not may_reconnect:
                 self.daemon.opts["dev-no-reconnect"] = None
@@ -595,7 +596,7 @@ def __init__(self, node_id, lightning_dir, bitcoind, executor, may_fail=False,
         dsn = db.get_dsn()
         if dsn is not None:
             self.daemon.opts['wallet'] = dsn
-        if VALGRIND:
+        if valgrind:
             self.daemon.cmd_prefix = [
                 'valgrind',
                 '-q',
@@ -958,7 +959,11 @@ def passes_filters(hmsg, filters):
 class NodeFactory(object):
     """A factory to setup and start `lightningd` daemons.
     """
-    def __init__(self, testname, bitcoind, executor, directory, db_provider, node_cls):
+    def __init__(self, request, testname, bitcoind, executor, directory, db_provider, node_cls):
+        if request.node.get_closest_marker("slow_test") and SLOW_MACHINE:
+            self.valgrind = False
+        else:
+            self.valgrind = VALGRIND
         self.testname = testname
         self.next_id = 1
         self.nodes = []
@@ -1042,7 +1047,7 @@ def get_node(self, node_id=None, options=None, dbfile=None,
         # node.
         db = self.db_provider.get_db(os.path.join(lightning_dir, TEST_NETWORK), self.testname, node_id)
         node = self.node_cls(
-            node_id, lightning_dir, self.bitcoind, self.executor, db=db,
+            node_id, lightning_dir, self.bitcoind, self.executor, self.valgrind, db=db,
             port=port, options=options, may_fail=may_fail or expect_fail,
             **kwargs
         )
@@ -1072,13 +1077,10 @@ def get_node(self, node_id=None, options=None, dbfile=None,
                 raise
         return node
 
-    def line_graph(self, num_nodes, fundchannel=True, fundamount=10**6, wait_for_announce=False, opts=None, announce_channels=True):
-        """ Create nodes, connect them and optionally fund channels.
-        """
+    def join_nodes(self, nodes, fundchannel=True, fundamount=10**6, wait_for_announce=False, announce_channels=True) -> None:
+        """Given nodes, connect them in a line, optionally funding a channel"""
         assert not (wait_for_announce and not announce_channels), "You've asked to wait for an announcement that's not coming. (wait_for_announce=True,announce_channels=False)"
-        nodes = self.get_nodes(num_nodes, opts=opts)
-        bitcoin = nodes[0].bitcoin
-        connections = [(nodes[i], nodes[i + 1]) for i in range(0, num_nodes - 1)]
+        connections = [(nodes[i], nodes[i + 1]) for i in range(len(nodes) - 1)]
 
         for src, dst in connections:
             src.rpc.connect(dst.info['id'], 'localhost', dst.port)
@@ -1088,32 +1090,55 @@ def line_graph(self, num_nodes, fundchannel=True, fundamount=10**6, wait_for_ann
         if not fundchannel:
             for src, dst in connections:
                 dst.daemon.wait_for_log(r'{}-.*openingd-chan#[0-9]*: Handed peer, entering loop'.format(src.info['id']))
-            return nodes
+            return
 
+        bitcoind = nodes[0].bitcoin
         # If we got here, we want to fund channels
         for src, dst in connections:
             addr = src.rpc.newaddr()['bech32']
-            src.bitcoin.rpc.sendtoaddress(addr, (fundamount + 1000000) / 10**8)
+            bitcoind.rpc.sendtoaddress(addr, (fundamount + 1000000) / 10**8)
 
-        bitcoin.generate_block(1)
+        bitcoind.generate_block(1)
+        sync_blockheight(bitcoind, nodes)
+        txids = []
         for src, dst in connections:
-            wait_for(lambda: len(src.rpc.listfunds()['outputs']) > 0)
-            tx = src.rpc.fundchannel(dst.info['id'], fundamount, announce=announce_channels)
-            wait_for(lambda: tx['txid'] in bitcoin.rpc.getrawmempool())
+            txids.append(src.rpc.fundchannel(dst.info['id'], fundamount, announce=announce_channels)['txid'])
+
+        wait_for(lambda: set(txids).issubset(set(bitcoind.rpc.getrawmempool())))
 
         # Confirm all channels and wait for them to become usable
-        bitcoin.generate_block(1)
+        bitcoind.generate_block(1)
         scids = []
         for src, dst in connections:
             wait_for(lambda: src.channel_state(dst) == 'CHANNELD_NORMAL')
             scid = src.get_channel_scid(dst)
-            src.daemon.wait_for_log(r'Received channel_update for channel {scid}/. now ACTIVE'.format(scid=scid))
             scids.append(scid)
 
+        # We don't want to assume message order here.
+        # Wait for ends:
+        nodes[0].daemon.wait_for_logs([r'update for channel {}/0 now ACTIVE'
+                                       .format(scids[0]),
+                                       r'update for channel {}/1 now ACTIVE'
+                                       .format(scids[0])])
+        nodes[-1].daemon.wait_for_logs([r'update for channel {}/0 now ACTIVE'
+                                        .format(scids[-1]),
+                                        r'update for channel {}/1 now ACTIVE'
+                                        .format(scids[-1])])
+        # Now wait for intermediate nodes:
+        for i, n in enumerate(nodes[1:-1]):
+            n.daemon.wait_for_logs([r'update for channel {}/0 now ACTIVE'
+                                    .format(scids[i]),
+                                    r'update for channel {}/1 now ACTIVE'
+                                    .format(scids[i]),
+                                    r'update for channel {}/0 now ACTIVE'
+                                    .format(scids[i + 1]),
+                                    r'update for channel {}/1 now ACTIVE'
+                                    .format(scids[i + 1])])
+
         if not wait_for_announce:
-            return nodes
+            return
 
-        bitcoin.generate_block(5)
+        bitcoind.generate_block(5)
 
         def both_dirs_ready(n, scid):
             resp = n.rpc.listchannels(scid)
@@ -1129,6 +1154,12 @@ def both_dirs_ready(n, scid):
             for end in (nodes[0], nodes[-1]):
                 wait_for(lambda: 'alias' in only_one(end.rpc.listnodes(n.info['id'])['nodes']))
 
+    def line_graph(self, num_nodes, fundchannel=True, fundamount=10**6, wait_for_announce=False, opts=None, announce_channels=True):
+        """ Create nodes, connect them and optionally fund channels.
+        """
+        nodes = self.get_nodes(num_nodes, opts=opts)
+
+        self.join_nodes(nodes, fundchannel, fundamount, wait_for_announce, announce_channels)
         return nodes
 
     def killall(self, expected_successes):
@@ -1139,7 +1170,7 @@ def killall(self, expected_successes):
             leaks = None
             # leak detection upsets VALGRIND by reading uninitialized mem.
             # If it's dead, we'll catch it below.
-            if not VALGRIND and DEVELOPER:
+            if not self.valgrind and DEVELOPER:
                 try:
                     # This also puts leaks in log.
                     leaks = self.nodes[i].rpc.dev_memleak()['leaks']