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 for parallel test execution.

The ClusterGetter class is responsible for managing a pool of cluster instances and assigning them to tests running in parallel on different pytest workers. It ensures that tests get a suitable, properly configured, and healthy cluster instance to run on.

Coordination between workers is achieved through status records kept in a SQLite database in a shared temporary directory. The records signal the state of each cluster instance (e.g., running, needs respin), which tests are running on which instance, and what resources are locked or in use.

The core logic is implemented in the get_cluster_instance method. It enters a loop where it evaluates the state of all available cluster instances against the requirements of the current test (e.g., resource needs, custom scripts, priority). It will wait and retry until a suitable instance is found and all conditions for starting the test are met. This includes handling cluster restarts (respins), resource allocation, and synchronization for tests that share expensive setups (marked tests).

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 | ResourceFilter] = (), use_resources: Iterable[str | ResourceFilter] = (), prio: bool = False, cleanup: bool = False, scriptsdir: str | Path = '') → 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. Mark the tests also with @pytest.mark.xdist_group(“<same name>”) so they are scheduled together on a single pytest worker - the mark is considered abandoned and cleaned up (see MARK_STALENESS_SEC) when no marked test is running for too long. Where applicable, prefer subtests (pytest_subtests) over mark - a single test with multiple subtests shares one setup with less magic. Subtests however don’t work with hypothesis property based tests.

• 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).

• scriptsdir – Path to custom scripts for the cluster.

property instance_dir: Path

property ports: InstancePorts

Return port mappings for current cluster instance.

property snap: StatusSnapshot

Return the snapshot of status records.

The snapshot is valid only while the global cluster lock is held and must be refreshed right after the lock is acquired.

cardano_node_tests.cluster_management.cluster_management module

Module for exposing useful components of cluster management.

The cluster management system is designed to manage a pool of Cardano cluster instances for running tests in parallel using pytest-xdist. It coordinates access to these shared cluster instances by multiple test workers.

Key concepts:

• Pool of Instances: Multiple cluster instances can be running concurrently. Each test worker requests a cluster instance to run a test on.

• Coordination via Shared Database: Workers communicate and coordinate through status records kept in a SQLite database on a shared file system. The records act as signals that indicate the state of cluster instances (e.g., which test is running, if a respin is needed, which resources are locked). The status_db module manages the creation and lookup of these records.

• Resource Management: Tests can declare what resources they need. A resource can be, for example, a specific feature of a cluster that cannot be used by multiple tests at the same time. The ClusterManager handles locking of these resources so that only one test can use them at a time.

• Cluster Respin: Some tests can modify the state of a cluster in a way that it cannot be used by subsequent tests. These tests can request a “respin” of the cluster instance, which re-initializes it to a clean state.

• `ClusterManager`: This is the main class that test fixtures interact with. Its get() method is used to acquire a suitable cluster instance for a test, taking into account available instances, resource requirements, and scheduling priority.

This system allows for efficient parallel execution of tests that require a running Cardano cluster, by reusing cluster instances and managing contention for shared resources.

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

High-level management of cluster instances.

This module provides the ClusterManager class, which is the main interface for tests to get a fully initialized cluster instance. The ClusterManager is responsible for selecting an available cluster instance that meets the test’s resource requirements, preparing the clusterlib object, and performing cleanup actions after the test has finished.

The ClusterManager is instantiated by the cluster_manager fixture for each test worker and is used by the cluster fixture to get a cluster instance for a test.

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 | ResourceFilter] = (), use_resources: Iterable[str | ResourceFilter] = (), prio: bool = False, cleanup: bool = False, scriptsdir: str | Path = '', 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.

Use worker_id=”*” to get resources locked by any worker.

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.

Use worker_id=”*” to get resources used by any worker.

init(mark: str = '', lock_resources: Iterable[str | ResourceFilter] = (), use_resources: Iterable[str | ResourceFilter] = (), prio: bool = False, cleanup: bool = False, scriptsdir: str | Path = '') → 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

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_management module

Functionality for selecting a cluster instance that has required resources available.

Resources can be requested by name (string).

It is also possible to use filters. A filter is an object that gets a list of all unavailable resources and returns a list of resources that should be used. An example is OneOf, which returns one usable resource from a given list of resources. The unavailable resources passed to the filter include resources that are locked, resources that were requested by name in the same request, and resources that were already selected by preceding filters in the same request.

It is possible to use multiple OneOf filters in a single request. For example, using OneOf filter with the same set of resources twice will result in selecting two different resources from that set.

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

Bases: ResourceFilter

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

filter(unavailable: Iterable[str]) → list[str]

Return one randomly selected usable resource, or an empty list if none is usable.

class cardano_node_tests.cluster_management.resources_management.ResourceFilter(*args, **kwargs)

Bases: Protocol

Protocol for resource filters.

filter(unavailable: Iterable[str]) → list[str]

Return resources to use, given the currently unavailable resources.

The unavailable resources include resources that are locked, resources that were requested by name in the same request, and resources that were already selected by preceding filters in the same request.

Returning an empty list means the request cannot be satisfied.

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

Get resources that can be used or locked.

cardano_node_tests.cluster_management.status_db module

SQLite-backed store for cluster instances status.

The status records are used for communication and synchronization between pytest workers.

The SQLite database file is created in the single temp directory shared by all workers (the directory returned by temptools.get_pytest_root_tmp()). This allows all workers to see status records created by other workers.

The database is a pure state store. Mutual exclusion between workers is provided by the global cluster file lock (see cluster_management.common.CLUSTER_LOCK), the same way it was when the state was kept in status files. The database is still configured defensively (WAL journal, busy timeout) so concurrent readers - e.g. humans inspecting the database - never block the workers.

Filter arguments semantics used throughout this module:

• instance_num: None matches any cluster instance, an int matches exactly.

• worker_id: “*” matches any pytest worker, any other string matches exactly.

• mark: None matches any record (with or without mark), “*” matches any non-empty mark, “” matches only records without mark, any other string matches exactly.

The current status can be inspected with the stock sqlite3 CLI. The overview view combines all status records into one human-readable table:

db=/tmp/pytest-of-$USER/pytest-0/cm-status.db sqlite3 -readonly -header “$db” ‘SELECT * FROM overview ORDER BY instance_num, kind’

The underlying tables (test_running, resources, flags) can be queried directly the same way.

class cardano_node_tests.cluster_management.status_db.StatusRow(instance_num: int, worker_id: str, mark: str, test_id: str = '', name: str = '', created_at: float = 0.0)

Bases: object

Single status record.

The test_id field is set only for “test running” records and the name field is set only for resource records.

created_at: float = 0.0

instance_num: int

mark: str

name: str = ''

test_id: str = ''

worker_id: str

class cardano_node_tests.cluster_management.status_db.StatusSnapshot

Bases: object

In-memory view of all status records.

The snapshot is loaded with one query per table and its read accessors mirror the module-level query functions, including the filter argument semantics. Reads from the snapshot are plain list traversals, so e.g. the scheduler can evaluate all cluster instances without doing repeated database queries.

The snapshot refreshes itself when this process modifies the database (tracked by the module-level write generation counter). Writes done by other processes are NOT visible until refresh is called explicitly. Users must therefore hold the global cluster lock while using the snapshot and must call refresh right after acquiring the lock.

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

Return list of marks of currently running tests.

get_resource_names(mode: str, instance_num: int | None = None, worker_id: str = '*', mark: str | None = None) → list[str]

Return names of resources that are locked or in use.

is_cluster_dead(instance_num: int) → bool

Check if the cluster instance is in broken state.

is_cluster_running(instance_num: int) → bool

Check if the cluster instance is running.

is_cluster_stopped(instance_num: int) → bool

Check if the cluster instance is stopped.

list_cluster_dead(instance_num: int | None = None) → list[StatusRow]

List all “cluster dead” records.

list_curr_mark(instance_num: int | None = None, worker_id: str = '*', mark: str = '*') → list[StatusRow]

List all “current mark” records.

list_prio_in_progress(worker_id: str = '*') → list[StatusRow]

List all “priority test in progress” records.

list_respin_needed(instance_num: int | None = None, worker_id: str = '*') → list[StatusRow]

List all “needs respin” records.

list_respin_progress(instance_num: int | None = None, worker_id: str = '*') → list[StatusRow]

List all “respin in progress” records.

list_test_running(instance_num: int | None = None, worker_id: str = '*', mark: str | None = None) → list[StatusRow]

List all “test running” records.

refresh() → None

Reload all status records from the database.

cardano_node_tests.cluster_management.status_db.create_curr_mark(instance_num: int, worker_id: str, mark: str) → None

Indicate presence of marked test on a pytest worker.

cardano_node_tests.cluster_management.status_db.create_prio_in_progress(worker_id: str) → None

Indicate that priority test is in progress.

cardano_node_tests.cluster_management.status_db.create_resources(instance_num: int, worker_id: str, names: Iterable[str], mode: str, mark: str = '') → None

Create records that indicate that the given resources are locked or in use.

Parameters:

• instance_num – Cluster instance number.

• worker_id – Pytest worker ID.

• names – Names of the resources.

• mode – Either MODE_LOCK or MODE_USE.

• mark – Test mark, empty string for no mark.

cardano_node_tests.cluster_management.status_db.create_respin_after_mark(instance_num: int, worker_id: str, mark: str) → None

Indicate that the cluster instance needs respin after marked tests are finished.

cardano_node_tests.cluster_management.status_db.create_respin_needed(instance_num: int, worker_id: str) → None

Indicate that the cluster instance needs respin.

cardano_node_tests.cluster_management.status_db.create_respin_progress(instance_num: int, worker_id: str) → None

Indicate that respin of the cluster instance is in progress.

cardano_node_tests.cluster_management.status_db.create_test_running(instance_num: int, worker_id: str, test_id: str, mark: str = '') → None

Indicate that a test is running on a pytest worker.

cardano_node_tests.cluster_management.status_db.gc_stale_records(min_interval_sec: float = 0.0) → list[str]

Remove status records left by pytest workers that are no longer running.

Must be called under the global cluster lock.

When min_interval_sec is set, the garbage collection runs at most once per that interval across all workers - the last-run timestamp is kept in the database. Throttled calls return an empty list without scanning.

Only records whose validity depends on the writer being alive are removed:

• “test running” records - the test is not running anymore.

• Resource records without mark - the test that held them is gone. Marked resource records belong to the whole group of marked tests and are cleaned up by the mark staleness handling, or removed together with the marks when the cluster instance is respun or failed to start.

• “priority test in progress” flags - would otherwise block all other workers forever.

• “respin in progress” flags - the cluster instance was left in an unknown state, so a “needs respin” flag is created in its place.

The “needs respin” and “respin after mark” flags are kept even when their writer died, as the cluster instance still needs the respin. The “current mark” flags are kept too - once the ghost “test running” records are removed, the existing mark staleness handling cleans them up.

Returns descriptions of the removed records, for logging.

cardano_node_tests.cluster_management.status_db.get_db_file() → PathLike[str]

Return path to the status database file.

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

Return list of marks of currently running tests.

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

Return names of resources that are locked or in use.

cardano_node_tests.cluster_management.status_db.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.

cardano_node_tests.cluster_management.status_db.is_cluster_dead(instance_num: int) → bool

Check if the cluster instance is in broken state.

cardano_node_tests.cluster_management.status_db.is_cluster_running(instance_num: int) → bool

Check if the cluster instance is running.

cardano_node_tests.cluster_management.status_db.is_cluster_stopped(instance_num: int) → bool

Check if the cluster instance is stopped.

cardano_node_tests.cluster_management.status_db.list_cluster_dead(instance_num: int | None = None) → list[StatusRow]

List all “cluster dead” records.

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

List all “current mark” records.

cardano_node_tests.cluster_management.status_db.list_prio_in_progress(worker_id: str = '*') → list[StatusRow]

List all “priority test in progress” records.

cardano_node_tests.cluster_management.status_db.list_resources(mode: str, instance_num: int | None = None, worker_id: str = '*', mark: str | None = None) → list[StatusRow]

List all resource records for the given mode.

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

List all “respin after mark” records.

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

List all “needs respin” records.

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

List all “respin in progress” records.

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

List all “test running” records.

cardano_node_tests.cluster_management.status_db.refresh_curr_mark(instance_num: int, mark: str) → None

Update creation time of “current mark” records to the current time.

Called whenever a test with the given mark is seen running or finishes, so the age of a “current mark” record tells for how long no marked test was running. All the mark’s records (one per worker) are refreshed together, as the mark is also cleaned up as a whole.

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

Delete all “current mark” records.

cardano_node_tests.cluster_management.status_db.rm_prio_in_progress(worker_id: str = '*') → list[StatusRow]

Delete all “priority test in progress” records.

cardano_node_tests.cluster_management.status_db.rm_resources(mode: str, instance_num: int | None = None, worker_id: str = '*', mark: str | None = None) → list[StatusRow]

Delete all resource records for the given mode.

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

Delete all “respin after mark” records.

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

Delete all “needs respin” records.

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

Delete all “respin in progress” records.

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

Delete all “test running” records.

cardano_node_tests.cluster_management.status_db.set_cluster_dead(instance_num: int) → None

Indicate that the cluster instance is in broken state.

cardano_node_tests.cluster_management.status_db.set_cluster_running(instance_num: int) → None

Indicate that the cluster instance is running.

cardano_node_tests.cluster_management.status_db.set_cluster_stopped(instance_num: int) → None

Indicate that the cluster instance is stopped.

cardano_node_tests.cluster_management.status_files module

Cluster instance status files.

Most of the status records used for communication and synchronization between pytest workers are kept in a SQLite database (see the status_db module). This module handles the few remaining file-based pieces:

• Cluster instance directories inside the single temp directory shared by all workers (the directory returned by temptools.get_pytest_root_tmp()).

• The “started by framework” status file that is created in the cluster instance state directory. Unlike the per-run status database, the state directory persists across pytest runs, so a later run can tell whether an existing cluster instance was started by the test framework.

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.get_instance_dir(instance_num: int) → Path

Return cluster instance directory for the given instance number.

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.

Module contents

Functionality for parallel execution of tests on multiple cluster instances.