refactor: make ClusterType a Protocol

cardano_node_tests/cluster_management/cluster_getter.py

@@ -120,7 +120,7 @@ def __init__(
         self.pytest_tmp_dir = temptools.get_pytest_root_tmp()
         self.cluster_lock = common.get_cluster_lock_file()
 
-        if cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.LOCAL:
+        if cluster_nodes.get_cluster_type().is_local:
             # Soft timeout (seconds): applies when no cluster is selected.
             self.grace_period_soft = 3600
             # Hard timeout (seconds): always applies, regardless of cluster selection.

cardano_node_tests/tests/common.py

@@ -163,12 +163,12 @@ class XdSplits(enum.StrEnum):
 )
 
 SKIPIF_ON_TESTNET = pytest.mark.skipif(
-    cluster_nodes.get_cluster_type().type != cluster_nodes.ClusterType.LOCAL,
+    not cluster_nodes.get_cluster_type().is_local,
     reason="not supposed to run on long-running testnet",
 )
 
 SKIPIF_ON_LOCAL = pytest.mark.skipif(
-    cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.LOCAL,
+    cluster_nodes.get_cluster_type().is_local,
     reason="supposed to run on long-running testnet",
 )
 
@@ -239,7 +239,7 @@ class XdSplits(enum.StrEnum):
 
 
 # Intervals for `wait_for_epoch_interval` (negative values are counted from the end of an epoch)
-if cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.LOCAL:
+if cluster_nodes.get_cluster_type().is_local:
     # Time buffer at the end of an epoch, enough to do something that takes several transactions
     EPOCH_STOP_SEC_BUFFER = -40
     # Time when all ledger state info is available for the current epoch
@@ -400,7 +400,7 @@ def detect_fork(
 
     # Forked nodes are the ones that differ from the majority of nodes
     if forked_nodes and len(forked_nodes) > (len(known_nodes) // 2):
-        forked_nodes = known_nodes - forked_nodes
+        forked_nodes = set(known_nodes - forked_nodes)
 
     return forked_nodes, unsynced_nodes
 
@@ -677,7 +677,7 @@ def is_fee_in_interval(fee: float, expected_fee: float, frac: float = 0.1) -> bo
     range.
     """
     # We have the fees calibrated only for local testnet
-    if cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.TESTNET:
+    if cluster_nodes.get_cluster_type().is_testnet:
         return True
     return helpers.is_in_interval(fee, expected_fee, frac=frac)
 

cardano_node_tests/tests/conftest.py

@@ -251,7 +251,7 @@ def _stop_all_cluster_instances(cluster_manager_obj: cluster_management.ClusterM
 
 def _testnet_cleanup(pytest_root_tmp: pl.Path) -> None:
     """Perform testnet cleanup at the end of session."""
-    if cluster_nodes.get_cluster_type().type != cluster_nodes.ClusterType.TESTNET:
+    if not cluster_nodes.get_cluster_type().is_testnet:
         return
 
     # There's only one cluster instance for testnets, so we don't need to use cluster manager
@@ -411,7 +411,7 @@ def respin_on_large_db(
     if (
         os.environ.get("GITHUB_ACTIONS")
         and configuration.HAS_DBSYNC
-        and cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.LOCAL
+        and cluster_nodes.get_cluster_type().is_local
         and cluster_manager._cluster_instance_num != -1
     ):
         db_size = dbsync_queries.query_db_size()

cardano_node_tests/tests/delegation.py

@@ -81,7 +81,7 @@ def cluster_and_pool(
     single fixture.
     """
     cluster_type = cluster_nodes.get_cluster_type()
-    if cluster_type.type == cluster_nodes.ClusterType.TESTNET:
+    if cluster_type.is_testnet:
         cluster_obj: clusterlib.ClusterLib = cluster_manager.get(use_resources=use_resources)
 
         # Getting ledger state on official testnet is too expensive,

cardano_node_tests/tests/plutus_common.py

@@ -649,7 +649,7 @@ def check_plutus_costs(
 
     units: the time is in picoseconds and the space is in bytes.
     """
-    if cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.TESTNET:
+    if cluster_nodes.get_cluster_type().is_testnet:
         # We have the costs calibrated only for local testnet
         return
 

cardano_node_tests/tests/test_cli.py

@@ -1217,7 +1217,7 @@ def _check_stake_snapshot(  # noqa: C901
         expected_pool_ids_mapping = {p: helpers.decode_bech32(bech32=p) for p in expected_pool_ids}
 
         def _dump_on_error():
-            if cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.LOCAL:
+            if cluster_nodes.get_cluster_type().is_local:
                 clusterlib_utils.save_ledger_state(
                     cluster_obj=cluster_obj, state_name=temp_template
                 )

cardano_node_tests/tests/test_mir_certs.py

@@ -102,10 +102,8 @@ def skip_on_hf_shortcut(
     cluster_pots: clusterlib.ClusterLib,  # noqa: ARG001
 ) -> None:
     """Skip test if HF shortcut is used."""
-    if (
-        cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.LOCAL
-        and cluster_nodes.get_cluster_type().uses_shortcut
-    ):
+    cluster_type = cluster_nodes.get_cluster_type()
+    if cluster_type.is_local and cluster_type.uses_shortcut:
         pytest.skip("MIR certs testing is not supported on local cluster with HF shortcut.")
 
 

cardano_node_tests/tests/tests_conway/test_drep.py

@@ -95,7 +95,7 @@ def get_custom_drep(
     caching_key: str,
 ) -> governance_utils.DRepRegistration:
     """Create a custom DRep and cache it."""
-    if cluster_nodes.get_cluster_type().type != cluster_nodes.ClusterType.LOCAL:
+    if not cluster_nodes.get_cluster_type().is_local:
         pytest.skip("runs only on local cluster")
 
     fixture_cache: cluster_management.FixtureCache[governance_utils.DRepRegistration | None]
@@ -1087,7 +1087,7 @@ def test_dreps_delegation(
         check_delegation = (
             build_method == clusterlib_utils.BuildMethods.BUILD
             and submit_method == submit_utils.SubmitMethods.CLI
-            and cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.LOCAL
+            and cluster_nodes.get_cluster_type().is_local
             and "smoke" not in request.config.getoption("-m")
         )
 
@@ -1302,7 +1302,7 @@ def test_dreps_and_spo_delegation(
         check_delegation = (
             build_method == clusterlib_utils.BuildMethods.BUILD
             and submit_method == submit_utils.SubmitMethods.CLI
-            and cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.LOCAL
+            and cluster_nodes.get_cluster_type().is_local
             and "smoke" not in request.config.getoption("-m")
         )
 

cardano_node_tests/tests/tests_conway/test_guardrails.py

@@ -187,7 +187,7 @@ def _enact_script_constitution():
         if cur_constitution.get("script") != constitution_script_hash:
             if conway_common.is_in_bootstrap(cluster_obj=cluster):
                 pytest.skip("Cannot run update constitution during bootstrap period.")
-            if cluster_nodes.get_cluster_type().type != cluster_nodes.ClusterType.LOCAL:
+            if not cluster_nodes.get_cluster_type().is_local:
                 pytest.skip("Cannot run update constitution on non-local testnet.")
 
             _url = helpers.get_vcs_link()

cardano_node_tests/utils/cluster_nodes.py

@@ -50,62 +50,84 @@ class Testnets(enum.StrEnum):
     mainnet = "mainnet"
 
 
-class ClusterType:
-    """Generic cluster type."""
-
-    LOCAL: tp.Final[str] = "local"
-    TESTNET: tp.Final[str] = "testnet"
-    test_addr_records: tp.ClassVar[tuple[str, ...]] = (
-        "user1",
-        "user2",
-        "user3",
-        "user4",
-        "user5",
-    )
+class ClusterKind(enum.StrEnum):
+    LOCAL = "local"
+    TESTNET = "testnet"
+
+
+TEST_ADDR_RECORDS: tp.Final[tuple[str, ...]] = (
+    "user1",
+    "user2",
+    "user3",
+    "user4",
+    "user5",
+)
+
+# The message is a module-level constant so that each abstract protocol method body stays
+# a single `raise` statement and type checkers keep treating the methods as abstract (see
+# the same pattern in `cluster_scripts`).
+_NOT_IMPLEMENTED_MSG: tp.Final[str] = "Not implemented for this cluster type."
+
 
-    NODES: tp.ClassVar[set[str]] = set()
+class ClusterType(tp.Protocol):
+    """Protocol for cluster types."""
 
+    NODES: tp.ClassVar[frozenset[str]]
+
+    type: ClusterKind
     cluster_scripts: cluster_scripts.ScriptsTypes
 
-    def __init__(self) -> None:
-        self.type = "unknown"
+    @property
+    def is_local(self) -> bool:
+        """Check if the cluster runs on a local testnet."""
+        return self.type is ClusterKind.LOCAL
+
+    @property
+    def is_testnet(self) -> bool:
+        """Check if the cluster runs on a long-running public network (preview, mainnet, etc.)."""
+        return self.type is ClusterKind.TESTNET
 
     @property
     def testnet_type(self) -> str:
-        return ""
+        """Return testnet type (preview, preprod, etc.).
+
+        Returns an empty string on local cluster and "unknown" when the testnet is not
+        recognized.
+        """
+        raise NotImplementedError(_NOT_IMPLEMENTED_MSG)
 
     @property
     def uses_shortcut(self) -> bool:
         """Check if cluster uses shortcut to go from Byron to last supported era."""
-        msg = f"Not implemented for cluster type '{self.type}'."
-        raise NotImplementedError(msg)
+        raise NotImplementedError(_NOT_IMPLEMENTED_MSG)
 
     def get_cluster_obj(self, *, command_era: str = "") -> clusterlib.ClusterLib:
         """Return instance of `ClusterLib` (cluster_obj)."""
-        msg = f"Not implemented for cluster type '{self.type}'."
-        raise NotImplementedError(msg)
+        raise NotImplementedError(_NOT_IMPLEMENTED_MSG)
 
     def create_addrs_data(
         self, *, cluster_obj: clusterlib.ClusterLib, destination_dir: clusterlib.FileType = "."
     ) -> dict[str, dict[str, tp.Any]]:
         """Create addresses and their keys for usage in tests."""
-        msg = f"Not implemented for cluster type '{self.type}'."
-        raise NotImplementedError(msg)
+        raise NotImplementedError(_NOT_IMPLEMENTED_MSG)
 
 
 class LocalCluster(ClusterType):
     """Local cluster type (full cardano mode)."""
 
-    NODES: tp.ClassVar[set[str]] = {
-        "bft1",
-        *(f"pool{i}" for i in range(1, configuration.NUM_POOLS + 1)),
-    }
+    NODES: tp.ClassVar[frozenset[str]] = frozenset(
+        {"bft1", *(f"pool{i}" for i in range(1, configuration.NUM_POOLS + 1))}
+    )
 
     def __init__(self) -> None:
-        super().__init__()
-        self.type = ClusterType.LOCAL
+        self.type = ClusterKind.LOCAL
         self.cluster_scripts = cluster_scripts.LocalScripts()
 
+    @property
+    def testnet_type(self) -> str:
+        """Return empty string, local cluster is not a testnet."""
+        return ""
+
     @property
     def uses_shortcut(self) -> bool:
         """Check if cluster uses shortcut to go from Byron to last supported era."""
@@ -143,7 +165,7 @@ def create_addrs_data(
 
         # Create new addresses
         new_addrs_data: dict[str, dict[str, tp.Any]] = {}
-        for addr_name in self.test_addr_records:
+        for addr_name in TEST_ADDR_RECORDS:
             addr_name_instance = f"{addr_name}_ci{instance_num}"
             payment = cluster_obj.g_address.gen_payment_addr_and_keys(
                 name=addr_name_instance,
@@ -177,7 +199,7 @@ def create_addrs_data(
         # Fund new addresses from faucet address
         LOGGER.debug("Funding created addresses.")
         to_fund = [d["payment"] for d in new_addrs_data.values()]
-        amount_per_address = 100_000_000_000_000 // len(self.test_addr_records)
+        amount_per_address = 100_000_000_000_000 // len(TEST_ADDR_RECORDS)
         faucet.fund_from_faucet(
             *to_fund,
             cluster_obj=cluster_obj,
@@ -200,11 +222,10 @@ class TestnetCluster(ClusterType):
         1666656000: {"type": Testnets.preview, "byron_epochs": 0},
     }
 
-    NODES: tp.ClassVar[set[str]] = {"relay1"}
+    NODES: tp.ClassVar[frozenset[str]] = frozenset({"relay1"})
 
     def __init__(self) -> None:
-        super().__init__()
-        self.type = ClusterType.TESTNET
+        self.type = ClusterKind.TESTNET
         self.cluster_scripts = cluster_scripts.TestnetScripts()
 
         # Cached values
@@ -217,7 +238,7 @@ def uses_shortcut(self) -> bool:
 
     @property
     def testnet_type(self) -> str:
-        """Return testnet type (shelley_qa, etc.)."""
+        """Return testnet type (preview, preprod, etc.)."""
         if self._testnet_type:
             return self._testnet_type
 
@@ -258,12 +279,12 @@ def create_addrs_data(
             skey_file=shelley_dir / "faucet.skey",
         )
         faucet_addrs_data: dict[str, dict[str, tp.Any]] = {
-            self.test_addr_records[1]: {"payment": faucet_rec}
+            TEST_ADDR_RECORDS[1]: {"payment": faucet_rec}
         }
 
         # Create new addresses
         new_addrs_data: dict[str, dict[str, tp.Any]] = {}
-        for addr_name in self.test_addr_records[1:]:
+        for addr_name in TEST_ADDR_RECORDS[1:]:
             payment = cluster_obj.g_address.gen_payment_addr_and_keys(
                 name=addr_name,
                 destination_dir=destination_dir,
@@ -278,11 +299,11 @@ def create_addrs_data(
         # Fund new addresses from faucet address
         LOGGER.debug("Funding created addresses.")
         to_fund = [d["payment"] for d in new_addrs_data.values()]
-        amount_per_address = faucet_balance // len(self.test_addr_records)
+        amount_per_address = faucet_balance // len(TEST_ADDR_RECORDS)
         faucet.fund_from_faucet(
             *to_fund,
             cluster_obj=cluster_obj,
-            faucet_data=faucet_addrs_data[self.test_addr_records[1]],
+            faucet_data=faucet_addrs_data[TEST_ADDR_RECORDS[1]],
             amount=amount_per_address,
             destination_dir=destination_dir,
             force=True,

cardano_node_tests/utils/governance_setup.py

@@ -260,7 +260,7 @@ def get_default_governance(
     *, cluster_manager: cluster_management.ClusterManager, cluster_obj: clusterlib.ClusterLib
 ) -> governance_utils.GovernanceRecords:
     """Get default governance data for CC members, DReps and SPOs."""
-    if cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.TESTNET:
+    if cluster_nodes.get_cluster_type().is_testnet:
         err = "Default governance is not available on testnets"
         raise ValueError(err)
 

cardano_node_tests/utils/logfiles.py

@@ -109,7 +109,7 @@ def _get_ignored_error_regexes() -> list[str]:
         # on GitHub runners.
         errors_ignored.append("TraceBlockFromFuture")
 
-    if cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.TESTNET:
+    if cluster_nodes.get_cluster_type().is_testnet:
         errors_ignored.extend(
             (
                 # We can get this error when some clients are old, or are using wrong