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

property ports: InstancePorts

Return port mappings for current cluster instance.

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_cluster_lock_file() → str

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[Any]]

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.

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: T | None)

Bases: Generic[T]

Cache for a fixture.

value: T | None

cardano_node_tests.cluster_management.netstat_tools module

Functions based on netstat.

cardano_node_tests.cluster_management.netstat_tools.get_netstat_conn() → str

Get listing of connections from the netstat command.

cardano_node_tests.cluster_management.netstat_tools.get_netstat_listen() → str

Get listing of listening services from the netstat command.

cardano_node_tests.cluster_management.netstat_tools.kill_old_cluster(instance_num: int, log_func: Callable[[str], None]) → None

Attempt to kill all processes left over from a previous cluster instance.

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'

PLUTUS: Final[str] = 'plutus'

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.

cardano_node_tests.cluster_management.status_files module

cardano_node_tests.cluster_management.status_files.create_cluster_dead_file(instance_num: int) → Path

Create the status file that indicates that the cluster instance is in broken state.

cardano_node_tests.cluster_management.status_files.create_cluster_stopped_file(instance_num: int) → Path

Create the status file that indicates that the cluster instance is stopped.

cardano_node_tests.cluster_management.status_files.create_curr_mark_file(instance_num: int, worker_id: str, mark: str) → Path

Create the status file that indicates presence of marked test on a pytest worker.

cardano_node_tests.cluster_management.status_files.create_prio_in_progress_file(worker_id: str) → Path

Create the status file that indicates that priority test is in progress.

cardano_node_tests.cluster_management.status_files.create_resource_locked_files(instance_num: int, worker_id: str, lock_names: Iterable[str], mark: str = '') → list[Path]

Create status files that indicate that the given resources are locked.

cardano_node_tests.cluster_management.status_files.create_resource_used_files(instance_num: int, worker_id: str, use_names: Iterable[str], mark: str = '') → list[Path]

Create status files that indicate that the given resources are in use.

cardano_node_tests.cluster_management.status_files.create_respin_after_mark_file(instance_num: int, worker_id: str, mark: str) → Path

Create the status file that indicates that the cluster instance needs respin.

The respin will happen after marked tests are finished on the dedicated cluster instance.

cardano_node_tests.cluster_management.status_files.create_respin_needed_file(instance_num: int, worker_id: str) → Path

Create the status file that indicates that the cluster instance needs respin.

cardano_node_tests.cluster_management.status_files.create_respin_progress_file(instance_num: int, worker_id: str) → Path

Create the status file that indicates that respin is in progress.

cardano_node_tests.cluster_management.status_files.create_started_by_framework_file(state_dir: Path) → Path

Create the status file that indicates the cluster instance was started by test framework.

cardano_node_tests.cluster_management.status_files.create_test_running_file(instance_num: int, worker_id: str, test_id: str, mark: str = '') → Path

Create the status file that indicates that a test is running on a pytest worker.

Save the test name in the status file.

cardano_node_tests.cluster_management.status_files.get_cluster_dead_file(instance_num: int) → Path

Return the status file that indicates that the cluster instance is in broken state.

cardano_node_tests.cluster_management.status_files.get_cluster_running_file(instance_num: int) → Path

Return the status file that indicates that the cluster instance is running.

cardano_node_tests.cluster_management.status_files.get_cluster_stopped_file(instance_num: int) → Path

Return the status file that indicates that the cluster instance is stopped.

cardano_node_tests.cluster_management.status_files.get_curr_mark_file(instance_num: int, worker_id: str, mark: str) → Path

Return the status file that indicates presence of marked test on a pytest worker.

cardano_node_tests.cluster_management.status_files.get_instance_dir(instance_num: int) → Path

Return cluster instance directory for the given instance number.

cardano_node_tests.cluster_management.status_files.get_marks_in_progress(instance_num: int | None = None, worker_id: str = '*') → list[str]

Return list of marks that are in progress.

cardano_node_tests.cluster_management.status_files.get_prio_in_progress_file(worker_id: str) → Path

Return the status file that indicates that priority test is in progress.

cardano_node_tests.cluster_management.status_files.get_resources_from_path(paths: Iterable[Path]) → list[str]

Get resources names from status files path.

cardano_node_tests.cluster_management.status_files.get_respin_after_mark_file(instance_num: int, worker_id: str, mark: str) → Path

Return the status file that indicates that the cluster instance needs respin.

The respin will happen after marked tests are finished on the dedicated cluster instance.

cardano_node_tests.cluster_management.status_files.get_respin_needed_file(instance_num: int, worker_id: str) → Path

Return the status file that indicates that the cluster instance needs respin.

cardano_node_tests.cluster_management.status_files.get_respin_progress_file(instance_num: int, worker_id: str) → Path

Return the status file that indicates that respin is in progress.

cardano_node_tests.cluster_management.status_files.get_started_by_framework_file(state_dir: Path) → Path

Return the status file that indicates the cluster instance was started by test framework.

cardano_node_tests.cluster_management.status_files.get_test_names(instance_num: int | None = None, worker_id: str = '*', mark: str | None = None) → list[str]

Return list of test names that are currently running.

If mark is None, list all status files regarless whether any mark is present or not. If mark is *, list all status files that have any mark. If mark is an empty string, list all status files that don’t have mark.

cardano_node_tests.cluster_management.status_files.get_test_running_file(instance_num: int, worker_id: str, mark: str = '') → Path

Return the status file that indicates that a test is running on a pytest worker.

cardano_node_tests.cluster_management.status_files.list_cluster_dead_files(instance_num: int | None = None) → list[Path]

List all “cluster dead” status files.

cardano_node_tests.cluster_management.status_files.list_curr_mark_files(instance_num: int | None = None, worker_id: str = '*', mark: str = '*') → list[Path]

List all “current mark” status files.

cardano_node_tests.cluster_management.status_files.list_prio_in_progress_files(worker_id: str = '*') → list[Path]

List all “priority test in progress” status files.

cardano_node_tests.cluster_management.status_files.list_resource_locked_files(instance_num: int | None = None, worker_id: str = '*', mark: str | None = None) → list[Path]

List all “resource locked” status files.

If mark is None, list all status files regarless whether any mark is present or not. If mark is *, list all status files that have any mark. If mark is an empty string, list all status files that don’t have mark.

cardano_node_tests.cluster_management.status_files.list_resource_used_files(instance_num: int | None = None, worker_id: str = '*', mark: str | None = None) → list[Path]

List all “resource used” status files.

If mark is None, list all status files regarless whether any mark is present or not. If mark is *, list all status files that have any mark. If mark is an empty string, list all status files that don’t have mark.

cardano_node_tests.cluster_management.status_files.list_respin_after_mark_files(instance_num: int | None = None, worker_id: str = '*', mark: str = '*') → list[Path]

List all “respin after mark” status files.

cardano_node_tests.cluster_management.status_files.list_respin_needed_files(instance_num: int | None = None, worker_id: str = '*') → list[Path]

List all “needs respin” status files.

cardano_node_tests.cluster_management.status_files.list_respin_progress_files(instance_num: int | None = None, worker_id: str = '*') → list[Path]

List all “respin in progress” status files.

cardano_node_tests.cluster_management.status_files.list_test_running_files(instance_num: int | None = None, worker_id: str = '*', mark: str | None = None) → list[Path]

List all “test running” status files.

If mark is None, list all status files regarless whether any mark is present or not. If mark is *, list all status files that have any mark. If mark is an empty string, list all status files that don’t have mark.

cardano_node_tests.cluster_management.status_files.rm_curr_mark_files(instance_num: int | None = None, worker_id: str = '*', mark: str = '*') → list[Path]

Delete all “current mark” status files.

cardano_node_tests.cluster_management.status_files.rm_prio_in_progress_files(worker_id: str = '*') → list[Path]

Delete all “priority test in progress” status files.

cardano_node_tests.cluster_management.status_files.rm_resource_locked_files(instance_num: int | None = None, worker_id: str = '*', mark: str | None = None) → list[Path]

Delete all “resource locked” status files.

If mark is None, list all status files regarless whether any mark is present or not. If mark is *, list all status files that have any mark. If mark is an empty string, list all status files that don’t have mark.

cardano_node_tests.cluster_management.status_files.rm_resource_used_files(instance_num: int | None = None, worker_id: str = '*', mark: str | None = None) → list[Path]

Delete all “resource used” status files.

If mark is None, list all status files regarless whether any mark is present or not. If mark is *, list all status files that have any mark. If mark is an empty string, list all status files that don’t have mark.

cardano_node_tests.cluster_management.status_files.rm_respin_after_mark_files(instance_num: int | None = None, worker_id: str = '*', mark: str = '*') → list[Path]

Delete all “respin after mark” status files.

cardano_node_tests.cluster_management.status_files.rm_respin_needed_files(instance_num: int | None = None, worker_id: str = '*') → list[Path]

Delete all “needs respin” status files.

cardano_node_tests.cluster_management.status_files.rm_respin_progress_files(instance_num: int | None = None, worker_id: str = '*') → list[Path]

Delete all “respin in progress” status files.

cardano_node_tests.cluster_management.status_files.rm_test_running_files(instance_num: int | None = None, worker_id: str = '*', mark: str | None = None) → list[Path]

Delete all “test running” status files.

If mark is None, list all status files regarless whether any mark is present or not. If mark is *, list all status files that have any mark. If mark is an empty string, list all status files that don’t have mark.

Module contents

Functionality for parallel execution of tests on multiple cluster instances.