Refactor cluster manager

* cleaner, commented and more readable code
* rework how cluster instances are stopped
* rework `DEV_CLUSTER_RUNNING` dev instance so it can be used for
  parallel execution of tests

cardano_node_tests/tests/conftest.py

@@ -195,35 +195,18 @@ def _save_env_for_allure(pytest_config: Config) -> None:
 def testenv_setup_teardown(
     tmp_path_factory: TempdirFactory, worker_id: str, request: FixtureRequest
 ) -> Generator[None, None, None]:
-    pytest_root_tmp = temptools.get_pytest_root_tmp(tmp_path_factory)
-
     # save environment info for Allure
     _save_env_for_allure(request.config)
 
-    if not worker_id or worker_id == "master":
-        # if cluster was started outside of test framework, do nothing
-        if cluster_management.DEV_CLUSTER_RUNNING:
-            # TODO: check that socket is open and print error if not
-            yield
-            return
-
+    # if cluster was started outside of test framework, do nothing
+    if configuration.DEV_CLUSTER_RUNNING:
+        # TODO: check that socket is open and print error if not
         yield
-
-        cluster_manager_obj = cluster_management.ClusterManager(
-            tmp_path_factory=tmp_path_factory, worker_id=worker_id, pytest_config=request.config
-        )
-        cluster_manager_obj.save_worker_cli_coverage()
-        _testnet_cleanup(pytest_root_tmp=pytest_root_tmp)
-        _stop_all_cluster_instances(
-            tmp_path_factory=tmp_path_factory,
-            worker_id=worker_id,
-            pytest_config=request.config,
-            pytest_tmp_dir=pytest_root_tmp,
-        )
         return
 
-    # pylint: disable=consider-using-with
-    open(pytest_root_tmp / f".started_session_{worker_id}", "a", encoding="utf-8").close()
+    pytest_root_tmp = temptools.get_pytest_root_tmp(tmp_path_factory)
+
+    helpers.touch(pytest_root_tmp / f".started_session_{worker_id}")
 
     yield
 
@@ -233,7 +216,7 @@ def testenv_setup_teardown(
         )
         cluster_manager_obj.save_worker_cli_coverage()
 
-        os.remove(pytest_root_tmp / f".started_session_{worker_id}")
+        (pytest_root_tmp / f".started_session_{worker_id}").unlink()
         if not list(pytest_root_tmp.glob(".started_session_*")):
             _testnet_cleanup(pytest_root_tmp=pytest_root_tmp)
             _stop_all_cluster_instances(

cardano_node_tests/tests/test_configuration.py

@@ -103,7 +103,9 @@ def cluster_epoch_length(
     cluster_manager: cluster_management.ClusterManager, epoch_length_start_cluster: Path
 ) -> clusterlib.ClusterLib:
     return cluster_manager.get(
-        singleton=True, cleanup=True, start_cmd=str(epoch_length_start_cluster)
+        lock_resources=[cluster_management.Resources.CLUSTER],
+        cleanup=True,
+        start_cmd=str(epoch_length_start_cluster),
     )
 
 
@@ -112,7 +114,9 @@ def cluster_slot_length(
     cluster_manager: cluster_management.ClusterManager, slot_length_start_cluster: Path
 ) -> clusterlib.ClusterLib:
     return cluster_manager.get(
-        singleton=True, cleanup=True, start_cmd=str(slot_length_start_cluster)
+        lock_resources=[cluster_management.Resources.CLUSTER],
+        cleanup=True,
+        start_cmd=str(slot_length_start_cluster),
     )
 
 

cardano_node_tests/tests/test_kes.py

@@ -2,7 +2,6 @@
 # pylint: disable=abstract-class-instantiated
 import json
 import logging
-import os
 import shutil
 import time
 from pathlib import Path
@@ -98,7 +97,11 @@ def short_kes_start_cluster(tmp_path_factory: TempdirFactory) -> Path:
 def cluster_kes(
     cluster_manager: cluster_management.ClusterManager, short_kes_start_cluster: Path
 ) -> clusterlib.ClusterLib:
-    return cluster_manager.get(singleton=True, cleanup=True, start_cmd=str(short_kes_start_cluster))
+    return cluster_manager.get(
+        lock_resources=[cluster_management.Resources.CLUSTER],
+        cleanup=True,
+        start_cmd=str(short_kes_start_cluster),
+    )
 
 
 class TestKES:
@@ -222,7 +225,7 @@ def _wait_epoch_chores(this_epoch: int):
                         ), f"The pool '{pool_name}' has produced blocks in epoch {this_epoch}"
 
             # generate new operational certificate with valid `--kes-period`
-            os.remove(opcert_file)
+            opcert_file.unlink()
             valid_opcert_file = cluster.gen_node_operational_cert(
                 node_name=node_name,
                 kes_vkey_file=pool_rec["kes_key_pair"].vkey_file,

cardano_node_tests/tests/test_update_proposals.py

@@ -45,7 +45,9 @@ def cluster_update_proposal(
         self,
         cluster_manager: cluster_management.ClusterManager,
     ) -> clusterlib.ClusterLib:
-        return cluster_manager.get(singleton=True, cleanup=True)
+        return cluster_manager.get(
+            lock_resources=[cluster_management.Resources.CLUSTER], cleanup=True
+        )
 
     @pytest.fixture
     def payment_addr(