test: wait for all quorum member connections

test/functional/test_framework/test_framework.py

@@ -21,6 +21,8 @@
 import sys
 import tempfile
 import time
+from types import SimpleNamespace
+import unittest
 from concurrent.futures import ThreadPoolExecutor
 
 from typing import List, Optional, Union
@@ -1996,20 +1998,24 @@ def check_sporks_same():
             return all(node.spork('show') == sporks for node in self.nodes[1:])
         self.wait_until(check_sporks_same, timeout=timeout, sleep=1)
 
-    def wait_for_quorum_connections(self, quorum_hash, expected_connections, mninfos, llmq_type_name="llmq_test", timeout = 60, wait_proc=None):
+    def wait_for_quorum_connections(self, quorum_hash, expected_connections, mninfos, llmq_type_name="llmq_test", timeout = 60, wait_proc=None, expected_member_count=1):
         def check_quorum_connections():
             def ret():
                 if wait_proc is not None:
                     wait_proc()
                 return False
 
+            matching_member_count = 0
             for mn in mninfos:
                 s = mn.get_node(self).quorum("dkgstatus")
                 for qs in s["session"]:
                     if qs["llmqType"] != llmq_type_name:
                         continue
                     if qs["status"]["quorumHash"] != quorum_hash:
                         continue
+                    matching_member_count += 1
+                    # -1 means no matching connections entry for this session
+                    connected = -1
                     for qc in s["quorumConnections"]:
                         if "quorumConnections" not in qc:
                             continue
@@ -2019,19 +2025,15 @@ def ret():
                             continue
                         if len(qc["quorumConnections"]) == 0:
                             continue
-                        cnt = 0
-                        for c in qc["quorumConnections"]:
-                            if c["connected"]:
-                                cnt += 1
-                        if cnt < expected_connections:
-                            return ret()
-                        return True
-                    # a session with no matching connections - not ok
-                    return ret()
-                # a node with no sessions - ok
-                pass
-            # no sessions at all - not ok
-            return ret()
+                        connected = sum(1 for c in qc["quorumConnections"] if c["connected"])
+                        break
+                    if connected < expected_connections:
+                        return ret()
+                    break
+                # a node with no matching session - ok (non-member)
+            if matching_member_count < expected_member_count:
+                return ret()
+            return True
 
         self.wait_until(check_quorum_connections, timeout=timeout, sleep=1)
 
@@ -2197,7 +2199,7 @@ def mine_quorum(self, llmq_type_name="llmq_test", llmq_type=100, expected_connec
         self.log.info("Expected quorum_hash:"+str(q))
         self.log.info("Waiting for phase 1 (init)")
         self.wait_for_quorum_phase(q, 1, expected_members, None, 0, mninfos_online, llmq_type_name=llmq_type_name)
-        self.wait_for_quorum_connections(q, expected_connections, mninfos_online, wait_proc=lambda: self.bump_mocktime(1), llmq_type_name=llmq_type_name)
+        self.wait_for_quorum_connections(q, expected_connections, mninfos_online, wait_proc=lambda: self.bump_mocktime(1), llmq_type_name=llmq_type_name, expected_member_count=expected_members)
         if spork23_active:
             self.wait_for_masternode_probes(q, mninfos_online, wait_proc=lambda: self.bump_mocktime(1))
 
@@ -2359,7 +2361,7 @@ def mine_cycle_quorum(self):
         self.log.info("quorumIndex 0: Waiting for phase 1 (init)")
         self.wait_for_quorum_phase(q_0, 1, expected_members, None, 0, mninfos_online, llmq_type_name)
         self.log.info("quorumIndex 0: Waiting for quorum connections (init)")
-        self.wait_for_quorum_connections(q_0, expected_connections, mninfos_online, llmq_type_name, wait_proc=lambda: self.bump_mocktime(1))
+        self.wait_for_quorum_connections(q_0, expected_connections, mninfos_online, llmq_type_name, wait_proc=lambda: self.bump_mocktime(1), expected_member_count=expected_members)
         if spork23_active:
             self.wait_for_masternode_probes(q_0, mninfos_online, wait_proc=lambda: self.bump_mocktime(1), llmq_type_name=llmq_type_name)
 
@@ -2374,7 +2376,7 @@ def mine_cycle_quorum(self):
         self.log.info("quorumIndex 1: Waiting for phase 1 (init)")
         self.wait_for_quorum_phase(q_1, 1, expected_members, None, 0, mninfos_online, llmq_type_name)
         self.log.info("quorumIndex 1: Waiting for quorum connections (init)")
-        self.wait_for_quorum_connections(q_1, expected_connections, mninfos_online, llmq_type_name, wait_proc=lambda: self.bump_mocktime(1))
+        self.wait_for_quorum_connections(q_1, expected_connections, mninfos_online, llmq_type_name, wait_proc=lambda: self.bump_mocktime(1), expected_member_count=expected_members)
         if spork23_active:
             self.wait_for_masternode_probes(q_1, mninfos_online, wait_proc=lambda: self.bump_mocktime(1), llmq_type_name=llmq_type_name)
 
@@ -2539,3 +2541,41 @@ def test():
                     c += 1
             return c >= count
         self.wait_until(test, timeout=timeout)
+
+
+class TestFrameworkQuorumConnections(unittest.TestCase):
+    def test_wait_for_all_members(self):
+        quorum_hash = "11" * 32
+
+        def mninfo(connected):
+            if connected is None:
+                status = {"session": [], "quorumConnections": []}
+            else:
+                status = {
+                    "session": [{
+                        "llmqType": "llmq_test",
+                        "status": {"quorumHash": quorum_hash},
+                    }],
+                    "quorumConnections": [{
+                        "llmqType": "llmq_test",
+                        "quorumHash": quorum_hash,
+                        "quorumConnections": [{"connected": value} for value in connected],
+                    }],
+                }
+            node = SimpleNamespace(quorum=lambda _command: status)
+            return SimpleNamespace(get_node=lambda _framework: node)
+
+        for name, connections, expected_ready, expected_wait_calls in [
+            ("incomplete member", [[True, True], [True, False]], False, 1),
+            ("all members ready", [[True, True], [True, True]], True, 0),
+            ("missing member session", [[True, True], None], False, 1),
+        ]:
+            with self.subTest(name):
+                results = []
+                framework = SimpleNamespace(wait_until=lambda predicate, **kwargs: results.append(predicate()))
+                wait_proc_calls = []
+                DashTestFramework.wait_for_quorum_connections(
+                    framework, quorum_hash, 2, [mninfo(c) for c in connections],
+                    wait_proc=lambda: wait_proc_calls.append(True), expected_member_count=2)
+                self.assertEqual(results, [expected_ready])
+                self.assertEqual(len(wait_proc_calls), expected_wait_calls)

test/functional/test_runner.py

@@ -84,6 +84,7 @@
     "crypto.ripemd160",
     "script",
     "segwit_addr",
+    "test_framework",
 ]
 
 EXTENDED_SCRIPTS = [