Don't overlap networks in e2e tests (#7698)

Author: eddyashton

tests/e2e_operations.py

@@ -173,69 +173,46 @@ def test_forced_ledger_chunk(network, args):
 @reqs.description("Forced snapshot")
 @app.scoped_txs()
 def test_forced_snapshot(network, args):
-    inner_args = copy.deepcopy(args)
-    inner_args.common_read_only_ledger_dir = (
-        None  # Side-effect setting which would break the starting node
-    )
-    inner_args.label = f"{inner_args.label}_forced_snapshot"
-    inner_args.snapshot_tx_interval = (
-        10000  # Large interval to avoid interference from regular snapshots
-    )
-
-    # Use a separate network to ensure unforced snapshots do not happen
-    with infra.network.network(
-        inner_args.nodes,
-        inner_args.binary_dir,
-        inner_args.debug_nodes,
-        pdb=inner_args.pdb,
-        txs=app.LoggingTxs("user0"),
-    ) as inner_network:
-        inner_network.start_and_open(inner_args)
-
-        primary, _ = inner_network.find_primary()
-
-        # Submit some dummy transactions
-        inner_network.txs.issue(inner_network, number_txs=3)
+    primary, _ = network.find_primary()
 
-        with primary.client() as c:
-            r = c.get("/node/commit").body.json()
-            hwm_pre_proposal = TxID.from_str(r["transaction_id"]).seqno
+    # Submit some dummy transactions
+    network.txs.issue(network, number_txs=3)
 
-        # Ensure there is at least one signature greater than the hwm
-        inner_network.txs.issue(inner_network, number_txs=1, wait_for_sync=True)
+    with primary.client() as c:
+        r = c.get("/node/commit").body.json()
+        hwm_pre_proposal = TxID.from_str(r["transaction_id"]).seqno
 
-        # Submit a proposal to force a snapshot
-        proposal_body, careful_vote = inner_network.consortium.make_proposal(
-            "trigger_snapshot"
-        )
-        proposal = inner_network.consortium.get_any_active_member().propose(
-            primary, proposal_body
-        )
-        proposal = inner_network.consortium.vote_using_majority(
-            primary,
-            proposal,
-            careful_vote,
-        )
+    # Ensure there is at least one signature greater than the hwm
+    network.txs.issue(network, number_txs=1, wait_for_sync=True)
 
-        # Issue some more transactions
-        inner_network.txs.issue(inner_network, number_txs=5)
+    # Submit a proposal to force a snapshot
+    proposal_body, careful_vote = network.consortium.make_proposal("trigger_snapshot")
+    proposal = network.consortium.get_any_active_member().propose(
+        primary, proposal_body
+    )
+    proposal = network.consortium.vote_using_majority(
+        primary,
+        proposal,
+        careful_vote,
+    )
 
-        snapshots_dir = inner_network.get_committed_snapshots(
-            primary, target_seqno=hwm_pre_proposal + 1
-        )
+    # Issue some more transactions
+    network.txs.issue(network, number_txs=5)
 
-        for s in os.listdir(snapshots_dir):
-            with ccf.ledger.Snapshot(os.path.join(snapshots_dir, s)) as snapshot:
-                snapshot_seqno = snapshot.get_public_domain().get_seqno()
-                if snapshot_seqno > hwm_pre_proposal:
-                    LOG.info(
-                        f"Found a snapshot at {snapshot_seqno} which is after the pre-proposal-high-water-mark {hwm_pre_proposal}"
-                    )
-                    return network
+    snapshots_dir = network.get_committed_snapshots(
+        primary, target_seqno=hwm_pre_proposal + 1
+    )
 
-        raise RuntimeError("Could not find matching snapshot file")
+    for s in os.listdir(snapshots_dir):
+        with ccf.ledger.Snapshot(os.path.join(snapshots_dir, s)) as snapshot:
+            snapshot_seqno = snapshot.get_public_domain().get_seqno()
+            if snapshot_seqno > hwm_pre_proposal:
+                LOG.info(
+                    f"Found a snapshot at {snapshot_seqno} which is after the pre-proposal-high-water-mark {hwm_pre_proposal}"
+                )
+                return
 
-    return network
+    raise RuntimeError("Could not find matching snapshot file")
 
 
 # https://github.com/microsoft/CCF/issues/1858
@@ -515,28 +492,32 @@ def test_snapshot_repr_digest(network, args):
         ), f"Expected partial body of {range_end + 1} bytes, got {len(r.body.data())}"
 
 
-def test_snapshot_selection(network, args):
-    inner_args = copy.deepcopy(args)
-    inner_args.common_read_only_ledger_dir = (
+def run_manual_snapshot_tests(const_args):
+    # Use a separate network with explicit args to ensure unforced snapshots do not happen
+
+    args = copy.deepcopy(const_args)
+    args.common_read_only_ledger_dir = (
         None  # Side-effect setting which would break the starting node
     )
-    inner_args.label = f"{inner_args.label}_snapshot_selection"
-    inner_args.snapshot_tx_interval = (
+    args.label = f"{args.label}_manual_snapshots"
+    args.snapshot_tx_interval = (
         10000  # Large interval to avoid interference from regular snapshots
     )
 
-    # Use a separate network to ensure unforced snapshots do not happen
     with infra.network.network(
-        inner_args.nodes,
-        inner_args.binary_dir,
-        inner_args.debug_nodes,
-        pdb=inner_args.pdb,
-    ) as inner_network:
-        inner_network.start_and_open(inner_args)
-        _test_snapshot_selection(inner_network, inner_args)
+        args.nodes,
+        args.binary_dir,
+        args.debug_nodes,
+        pdb=args.pdb,
+        txs=app.LoggingTxs("user0"),
+    ) as network:
+        network.start_and_open(args)
 
+        test_snapshot_selection(network, args)
+        test_forced_snapshot(network, args)
 
-def _test_snapshot_selection(network, args):
+
+def test_snapshot_selection(network, args):
     # Add nodes so we have at least 3
     while len(network.get_joined_nodes()) < 3:
         new_node = network.create_node()
@@ -656,6 +637,10 @@ def find_snapshot(node):
     for node in suspended:
         node.resume()
 
+    # Heal after all the suspensions, before running further tests
+    network.wait_for_new_primary_in((primary, *backups))
+    network.wait_for_node_commit_sync()
+
 
 def test_empty_snapshot(network, args):
 
@@ -1533,11 +1518,9 @@ def run_file_operations(args):
                 test_save_committed_ledger_files(network, args)
                 test_parse_snapshot_file(network, args)
                 test_forced_ledger_chunk(network, args)
-                test_forced_snapshot(network, args)
                 test_large_snapshot(network, args)
                 test_snapshot_access(network, args)
                 test_snapshot_repr_digest(network, args)
-                test_snapshot_selection(network, args)
                 test_empty_snapshot(network, args)
                 test_nulled_snapshot(network, args)
                 test_corrupt_snapshot_handling(network, args)
@@ -2914,6 +2897,7 @@ def run_snp_tests(args):
 def run(args):
     run_max_uncommitted_tx_count(args)
     run_file_operations(args)
+    run_manual_snapshot_tests(args)
     run_tls_san_checks(args)
     run_config_timeout_check(args)
     run_configuration_file_checks(args)

tests/infra/network.py

@@ -1782,13 +1782,14 @@ def wait_for_new_primary(
 
     def wait_for_new_primary_in(
         self,
-        expected_node_ids,
+        expected_nodes,
         nodes=None,
         timeout_multiplier=DEFAULT_TIMEOUT_MULTIPLIER,
     ):
         # We arbitrarily pick twice the election duration to protect ourselves against the somewhat
         # but not that rare cases when the first round of election fails (short timeout are particularly susceptible to this)
         timeout = self.observed_election_duration * timeout_multiplier
+        expected_node_ids = [n.node_id for n in expected_nodes]
         LOG.info(
             f"Waiting up to {timeout}s for a new primary in {expected_node_ids} to be elected..."
         )

tests/partitions_test.py

@@ -178,7 +178,7 @@ def test_isolate_primary_from_one_backup(network, args):
     rules.drop()
 
     # primary should now observe partitioned backup as primary
-    network.wait_for_new_primary_in({b_0.node_id}, nodes=[p])
+    network.wait_for_new_primary_in({b_0}, nodes=[p])
 
     LOG.info(f"Check that new primary {b_0.local_node_id} reports stable acks")
     last_ack = 0
@@ -884,7 +884,7 @@ def force_become_primary(network, args, target_node):
         network.wait_for_node_commit_sync(nodes=backups)
         rules.drop()
         # Wait for the old primary to observe the new one
-        network.wait_for_new_primary_in({target_node.node_id}, nodes=[primary])
+        network.wait_for_new_primary_in({target_node}, nodes=[primary])
         primary = target_node
 
     # Ensure a signature has been produced in the new term