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.

addrs_data: dict[source]
cluster_obj: ClusterLib | None = None[source]
last_checksum: str = ''[source]
test_data: dict[source]

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.

property cluster_instance_num: int[source]
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.

property instance_dir: Path[source]

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.common.get_resources_from_path(paths: Iterator[Path]) list[str][source]

Get resources names from status files path.

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.

property cluster_instance_num: int[source]
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_logfiles_errors() str[source]

Get errors found in cluster artifacts.

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 instance_dir: Path[source]
log(msg: str) None[source]

Log a message.

on_test_stop() None[source]

Perform actions after a test is finished.

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.

save_all_clusters_artifacts() None[source]

Save artifacts of all cluster instances.

save_worker_cli_coverage() None[source]

Save CLI coverage info collected by this pytest worker.

Must be done when session of the worker is about to finish, because there’s no other job to call _reload_cluster_obj and thus save CLI coverage of the old cluster_obj instance.

set_needs_respin() None[source]

Indicate that the cluster instance needs respin.

stop_all_clusters() None[source]

Stop all cluster instances.

class cardano_node_tests.cluster_management.manager.FixtureCache(value: Any)[source]

Bases: object

Cache for a fixture.

value: Any[source]

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.

ALL_POOLS: Final[tuple[str, ...]] = ('node-pool1', 'node-pool2', 'node-pool3')[source]
CLUSTER: Final[str] = 'cluster'[source]
COMMITTEE: Final[str] = 'committee'[source]
DREPS: Final[str] = 'dreps'[source]
PERF: Final[str] = 'performance'[source]
POOL1: Final[str] = 'node-pool1'[source]
POOL2: Final[str] = 'node-pool2'[source]
POOL3: Final[str] = 'node-pool3'[source]
POOL_FOR_OFFLINE: Final[str] = 'node-pool2'[source]
POTS: Final[tuple[str, ...]] = ('reserves', 'treasury', 'rewards')[source]
RESERVES: Final[str] = 'reserves'[source]
REWARDS: Final[str] = 'rewards'[source]
TREASURY: Final[str] = 'treasury'[source]
cardano_node_tests.cluster_management.resources.get_unsanitized(ls: Iterable[str]) list[str][source]

Return unsanitized resource names from the list.

cardano_node_tests.cluster_management.resources.sanitize_res_name(s: str) str[source]

Sanitize resource name to expected format.

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.

filter(unavailable: Iterable[str], **kwargs: Any) list[str][source]

Filter resources.

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.

filter(unavailable: Iterable[str], **kwargs: Any) list[str][source]

Filter 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.