cardano_node_tests.cluster_management package

Submodules

cardano_node_tests.cluster_management.cache module

class cardano_node_tests.cluster_management.cache.CacheManager

Bases: object

Set of cache management methods.

cache: ClassVar[Dict[int, ClusterManagerCache]] = {}

classmethod get_cache() → Dict[int, ClusterManagerCache]

classmethod get_instance_cache(instance_num: int) → ClusterManagerCache

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 = '')

Bases: object

Cache for a single cluster instance.

Here goes only data that makes sense to reuse in multiple tests.

addrs_data: dict

cluster_obj: ClusterLib | None = None

last_checksum: str = ''

test_data: dict

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)

Bases: object

Internal class that encapsulate functionality for getting a cluster instance.

property cluster_instance_num: int

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

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

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)

Bases: object

Set of management methods for cluster instances.

property cache: ClusterManagerCache

cache_fixture(key: str = '') → Iterator[FixtureCache]

Cache fixture value - context manager.

property cluster_instance_num: int

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

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]

Get resources locked by worker.

It is possible to use glob patterns for worker_id (e.g. worker_id=”*”).

get_logfiles_errors() → str

Get errors found in cluster artifacts.

get_used_resources(from_set: Iterable[str] | None = None, worker_id: str | None = None) → List[str]

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

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

log(msg: str) → None

Log a message.

on_test_stop() → None

Perform actions after a test is finished.

property ports: InstancePorts

Return port mappings for current cluster instance.

respin_on_failure() → Iterator[None]

Indicate that the cluster instance needs respin if command failed - context manager.

save_all_clusters_artifacts() → None

Save artifacts of all cluster instances.

save_worker_cli_coverage() → None

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

Indicate that the cluster instance needs respin.

stop_all_clusters() → None

Stop all cluster instances.

class cardano_node_tests.cluster_management.manager.FixtureCache(value: Any)

Bases: object

Cache for a fixture.

value: Any

cardano_node_tests.cluster_management.resources module

class cardano_node_tests.cluster_management.resources.Resources

Bases: object

Resources that can be used for lock_resources or use_resources.

ALL_POOLS: Final[Tuple[str, ...]] = ('node-pool1', 'node-pool2', 'node-pool3')

CLUSTER: Final[str] = 'cluster'

COMMITTEE: Final[str] = 'committee'

DREPS: Final[str] = 'dreps'

PERF: Final[str] = 'performance'

POOL1: Final[str] = 'node-pool1'

POOL2: Final[str] = 'node-pool2'

POOL3: Final[str] = 'node-pool3'

POOL_FOR_OFFLINE: Final[str] = 'node-pool2'

POTS: Final[Tuple[str, ...]] = ('reserves', 'treasury', 'rewards')

RESERVES: Final[str] = 'reserves'

REWARDS: Final[str] = 'rewards'

TREASURY: Final[str] = 'treasury'

cardano_node_tests.cluster_management.resources.get_unsanitized(ls: Iterable[str]) → List[str]

Return unsanitized resource names from the list.

cardano_node_tests.cluster_management.resources.sanitize_res_name(s: str) → str

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])

Bases: object

Base class for resource filters.

filter(unavailable: Iterable[str], **kwargs: Any) → List[str]

Filter resources.

class cardano_node_tests.cluster_management.resources_management.OneOf(resources: Iterable[str])

Bases: BaseFilter

Filter that returns one usable resource out of list of resources.

filter(unavailable: Iterable[str], **kwargs: Any) → List[str]

Filter resources.

cardano_node_tests.cluster_management.resources_management.get_resources(resources: Iterable[str | BaseFilter], unavailable: Iterable[str]) → List[str]

Get resources that can be used or locked.

Module contents

Functionality for parallel execution of tests on multiple cluster instances.