cardano_node_tests.cluster_management package
Submodules
cardano_node_tests.cluster_management.cache module
- class cardano_node_tests.cluster_management.cache.CacheManager[source]
Bases:
object
Set of cache management methods.
- cache: ClassVar[dict[int, ClusterManagerCache]] = {}[source]
- classmethod get_cache() dict[int, ClusterManagerCache] [source]
- classmethod get_instance_cache(instance_num: int) ClusterManagerCache [source]
- class cardano_node_tests.cluster_management.cache.ClusterManagerCache(cluster_obj: ~cardano_clusterlib.clusterlib_klass.ClusterLib | None = None, test_data: dict = <factory>, addrs_data: dict = <factory>, last_checksum: str = '')[source]
Bases:
object
Cache for a single cluster instance.
Here goes only data that makes sense to reuse in multiple tests.
cardano_node_tests.cluster_management.cluster_getter module
Functionality for obtaining and setting up a cluster instance.
- class cardano_node_tests.cluster_management.cluster_getter.ClusterGetter(worker_id: str, pytest_config: Config, num_of_instances: int, log_func: Callable)[source]
Bases:
object
Internal class that encapsulate functionality for getting a cluster instance.
- get_cluster_instance(mark: str = '', lock_resources: Iterable[str | BaseFilter] = (), use_resources: Iterable[str | BaseFilter] = (), prio: bool = False, cleanup: bool = False, start_cmd: str = '') int [source]
Return a number of initialized cluster instance once we can start the test.
It checks current conditions and waits if the conditions don’t allow to start the test right away.
- Parameters:
mark – A string marking group of tests. Useful when group of tests need the same expensive setup. The mark will make sure the marked tests run on the same cluster instance.
lock_resources – An iterable of resources (names of resources) that will be used exclusively by the test (or marked group of tests). A locked resource cannot be used by other tests.
use_resources – An iterable of resources (names of resources) that will be used by the test (or marked group of tests). The resources can be shared with other tests, however resources in use cannot be locked by other tests.
prio – A boolean indicating that the test has priority in obtaining cluster instance. All other tests that also want to get a cluster instance need to wait.
cleanup – A boolean indicating if the cluster will be respun after the test (or marked group of tests) is finished. Can be used only for tests that locked whole cluster (“singleton” tests).
start_cmd – Custom command to start the cluster.
cardano_node_tests.cluster_management.cluster_management module
Module for exposing useful components of cluster management.
cardano_node_tests.cluster_management.common module
cardano_node_tests.cluster_management.manager module
Functionality for managing cluster instances.
- class cardano_node_tests.cluster_management.manager.ClusterManager(worker_id: str, pytest_config: Config)[source]
Bases:
object
Set of management methods for cluster instances.
- property cache: ClusterManagerCache[source]
- cache_fixture(key: str = '') Iterator[FixtureCache] [source]
Cache fixture value - context manager.
- get(mark: str = '', lock_resources: Iterable[str | BaseFilter] = (), use_resources: Iterable[str | BaseFilter] = (), prio: bool = False, cleanup: bool = False, start_cmd: str = '', check_initialized: bool = True) ClusterLib [source]
Get cardano_clusterlib.ClusterLib object on an initialized cluster instance.
Convenience method that calls init.
- get_locked_resources(from_set: Iterable[str] | None = None, worker_id: str | None = None) list[str] [source]
Get resources locked by worker.
It is possible to use glob patterns for worker_id (e.g. worker_id=”*”).
- get_used_resources(from_set: Iterable[str] | None = None, worker_id: str | None = None) list[str] [source]
Get resources used by worker.
It is possible to use glob patterns for worker_id (e.g. worker_id=”*”).
- init(mark: str = '', lock_resources: Iterable[str | BaseFilter] = (), use_resources: Iterable[str | BaseFilter] = (), prio: bool = False, cleanup: bool = False, start_cmd: str = '') None [source]
Get an initialized cluster instance.
This method will wait until a cluster instance is ready to be used.
IMPORTANT: This method must be called before any other method of this class.
- property ports: InstancePorts[source]
Return port mappings for current cluster instance.
- respin_on_failure() Iterator[None] [source]
Indicate that the cluster instance needs respin if command failed - context manager.
cardano_node_tests.cluster_management.resources module
- class cardano_node_tests.cluster_management.resources.Resources[source]
Bases:
object
Resources that can be used for lock_resources or use_resources.
cardano_node_tests.cluster_management.resources_management module
Functionality for getting a cluster instance that has required resources available.
- class cardano_node_tests.cluster_management.resources_management.BaseFilter(resources: Iterable[str])[source]
Bases:
object
Base class for resource filters.
- class cardano_node_tests.cluster_management.resources_management.OneOf(resources: Iterable[str])[source]
Bases:
BaseFilter
Filter that returns one usable resource out of list of resources.
- cardano_node_tests.cluster_management.resources_management.get_resources(resources: Iterable[str | BaseFilter], unavailable: Iterable[str]) list[str] [source]
Get resources that can be used or locked.
Module contents
Functionality for parallel execution of tests on multiple cluster instances.