cardano_node_tests.utils package

Submodules

cardano_node_tests.utils.artifacts module

Functionality for collecting testing artifacts.

cardano_node_tests.utils.artifacts.copy_artifacts(pytest_tmp_dir: pathlib.Path, pytest_config: _pytest.config.Config) → None[source]: Copy collected tests and cluster artifacts to artifacts dir.

cardano_node_tests.utils.artifacts.save_cli_coverage(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, pytest_config: _pytest.config.Config) → Optional[pathlib.Path][source]: Save CLI coverage info.

cardano_node_tests.utils.artifacts.save_cluster_artifacts(save_dir: pathlib.Path, state_dir: pathlib.Path) → None[source]: Save cluster artifacts (logs, certs, etc.).

cardano_node_tests.utils.artifacts.save_start_script_coverage(log_file: pathlib.Path, pytest_config: _pytest.config.Config) → Optional[pathlib.Path][source]: Save info about CLI commands executed by cluster start script.

cardano_node_tests.utils.cluster_nodes module

Functionality for cluster setup and interaction with cluster nodes.

class cardano_node_tests.utils.cluster_nodes.ClusterEnv(socket_path, state_dir, work_dir, instance_num, cluster_era, tx_era)[source]

Bases: NamedTuple

cluster_era: str[source]: Alias for field number 4

instance_num: int[source]: Alias for field number 3

socket_path: pathlib.Path[source]: Alias for field number 0

state_dir: pathlib.Path[source]: Alias for field number 1

tx_era: str[source]: Alias for field number 5

work_dir: pathlib.Path[source]: Alias for field number 2

class cardano_node_tests.utils.cluster_nodes.ClusterType[source]

Bases: object

Generic cluster type.

LOCAL = 'local'[source]

NODES: Set[str] = {}[source]

TESTNET = 'testnet'[source]

TESTNET_NOPOOLS = 'testnet_nopools'[source]

create_addrs_data(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, destination_dir: Union[str, pathlib.Path] = '.') → Dict[str, Dict[str, Any]][source]: Create addresses and their keys for usage in tests.

get_cluster_obj(protocol: str = '', tx_era: str = '', slots_offset: int = 0) → cardano_clusterlib.clusterlib_klass.ClusterLib[source]: Return instance of ClusterLib (cluster_obj).

test_addr_records = ('user1',)[source]

property testnet_type: str[source]

property uses_shortcut: bool[source]: Check if cluster uses shortcut to go from Byron to last supported era.

class cardano_node_tests.utils.cluster_nodes.LocalCluster[source]

Bases: cardano_node_tests.utils.cluster_nodes.ClusterType

Local cluster type (full cardano mode).

NODES: Set[str] = {'bft1', 'pool1', 'pool2', 'pool3'}[source]

create_addrs_data(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, destination_dir: Union[str, pathlib.Path] = '.') → Dict[str, Dict[str, Any]][source]: Create addresses and their keys for usage in tests.

get_cluster_obj(protocol: str = '', tx_era: str = '', slots_offset: int = 0) → cardano_clusterlib.clusterlib_klass.ClusterLib[source]: Return instance of ClusterLib (cluster_obj).

property uses_shortcut: bool[source]: Check if cluster uses shortcut to go from Byron to last supported era.

class cardano_node_tests.utils.cluster_nodes.ServiceStatus(name, status, pid, uptime, message)[source]

Bases: NamedTuple

message: str[source]: Alias for field number 4

name: str[source]: Alias for field number 0

pid: Optional[int][source]: Alias for field number 2

status: str[source]: Alias for field number 1

uptime: Optional[str][source]: Alias for field number 3

class cardano_node_tests.utils.cluster_nodes.TestnetCluster[source]

Bases: cardano_node_tests.utils.cluster_nodes.ClusterType

Testnet cluster type (full cardano mode).

NODES: Set[str] = {'pool1', 'pool2', 'relay1'}[source]

TESTNETS: Dict[int, dict] = {1506203091: {'shelley_start': '2020-07-29T21:44:51Z', 'type': 'mainnet'}, 1506450213: {'shelley_start': '2020-08-01T18:23:33Z', 'type': 'staging'}, 1563999616: {'shelley_start': '2020-07-28T20:20:16Z', 'type': 'testnet'}, 1597669200: {'shelley_start': '2020-08-17T17:00:00Z', 'type': 'shelley_qa'}, 1654041600: {'byron_epochs': 4, 'type': 'preprod'}, 1666656000: {'byron_epochs': 0, 'type': 'preview'}}[source]

create_addrs_data(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, destination_dir: Union[str, pathlib.Path] = '.') → Dict[str, Dict[str, Any]][source]: Create addresses and their keys for usage in tests.

get_cluster_obj(protocol: str = '', tx_era: str = '', slots_offset: int = 0) → cardano_clusterlib.clusterlib_klass.ClusterLib[source]: Return instance of ClusterLib (cluster_obj).

property testnet_type: str[source]: Return testnet type (shelley_qa, etc.).

property uses_shortcut: bool[source]: Check if cluster uses shortcut to go from Byron to last supported era.

class cardano_node_tests.utils.cluster_nodes.TestnetNopoolsCluster[source]

Bases: cardano_node_tests.utils.cluster_nodes.TestnetCluster

Testnet cluster type (full cardano mode).

NODES: Set[str] = {'relay1'}[source]

class cardano_node_tests.utils.cluster_nodes.Testnets[source]

Bases: object

mainnet = 'mainnet'[source]

preprod = 'preprod'[source]

preview = 'preview'[source]

shelley_qa = 'shelley_qa'[source]

staging = 'staging'[source]

testnet = 'testnet'[source]

cardano_node_tests.utils.cluster_nodes.get_cardano_node_socket_path(instance_num: int, socket_file_name: str = '') → pathlib.Path[source]: Return path to socket file in the given cluster instance.

cardano_node_tests.utils.cluster_nodes.get_cluster_env() → cardano_node_tests.utils.cluster_nodes.ClusterEnv[source]: Get cardano cluster environment.

cardano_node_tests.utils.cluster_nodes.get_cluster_type() → cardano_node_tests.utils.cluster_nodes.ClusterType[source]: Return instance of the cluster type indicated by configuration.

cardano_node_tests.utils.cluster_nodes.get_instance_num() → int[source]: Get cardano cluster instance number.

cardano_node_tests.utils.cluster_nodes.load_addrs_data() → dict[source]: Load data about addresses and their keys for usage in tests.

cardano_node_tests.utils.cluster_nodes.load_pools_data(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib) → dict[source]: Load data for pools existing in the cluster environment.

cardano_node_tests.utils.cluster_nodes.restart_all_nodes(instance_num: Optional[int] = None) → None[source]: Restart all Cardano nodes of the running cluster.

cardano_node_tests.utils.cluster_nodes.restart_nodes(node_names: List[str], instance_num: Optional[int] = None) → None[source]: Restart list of Cardano nodes of the running cluster.

cardano_node_tests.utils.cluster_nodes.services_action(service_names: List[str], action: str, instance_num: Optional[int] = None) → None[source]: Perform action on services on the running cluster.

cardano_node_tests.utils.cluster_nodes.services_status(service_names: Optional[List[str]] = None, instance_num: Optional[int] = None) → List[cardano_node_tests.utils.cluster_nodes.ServiceStatus][source]: Return status info for list of services running on the running cluster (all by default).

cardano_node_tests.utils.cluster_nodes.set_cluster_env(instance_num: int, socket_file_name: str = '') → None[source]: Set env variables for the given cluster instance.

cardano_node_tests.utils.cluster_nodes.setup_test_addrs(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, destination_dir: Union[str, pathlib.Path] = '.') → pathlib.Path[source]: Set addresses and their keys up for usage in tests.

cardano_node_tests.utils.cluster_nodes.start_cluster(cmd: str, args: List[str]) → cardano_clusterlib.clusterlib_klass.ClusterLib[source]: Start cluster.

cardano_node_tests.utils.cluster_nodes.start_nodes(node_names: List[str], instance_num: Optional[int] = None) → None[source]: Start list of Cardano nodes of the running cluster.

cardano_node_tests.utils.cluster_nodes.stop_nodes(node_names: List[str], instance_num: Optional[int] = None) → None[source]: Stop list of Cardano nodes of the running cluster.

cardano_node_tests.utils.cluster_scripts module

Functionality for cluster scripts (starting and stopping clusters).

copying scripts and their configuration, so it can be atered by tests
setup of scripts and their configuration for starting of multiple cluster instances

class cardano_node_tests.utils.cluster_scripts.InstanceFiles(start_script, stop_script, start_script_args, dir)[source]

Bases: NamedTuple

dir: pathlib.Path[source]: Alias for field number 3

start_script: pathlib.Path[source]: Alias for field number 0

start_script_args: List[str][source]: Alias for field number 2

stop_script: pathlib.Path[source]: Alias for field number 1

class cardano_node_tests.utils.cluster_scripts.InstancePorts(base, webserver, submit_api, supervisor, relay1, ekg_relay1, prometheus_relay1, bft1, ekg_bft1, prometheus_bft1, pool1, ekg_pool1, prometheus_pool1, pool2, ekg_pool2, prometheus_pool2, pool3, ekg_pool3, prometheus_pool3, node_ports)[source]

Bases: NamedTuple

base: int[source]: Alias for field number 0

bft1: int[source]: Alias for field number 7

ekg_bft1: int[source]: Alias for field number 8

ekg_pool1: int[source]: Alias for field number 11

ekg_pool2: int[source]: Alias for field number 14

ekg_pool3: int[source]: Alias for field number 17

ekg_relay1: int[source]: Alias for field number 5

node_ports: Union[List[cardano_node_tests.utils.cluster_scripts.NodePorts], Tuple[()]][source]: Alias for field number 19

pool1: int[source]: Alias for field number 10

pool2: int[source]: Alias for field number 13

pool3: int[source]: Alias for field number 16

prometheus_bft1: int[source]: Alias for field number 9

prometheus_pool1: int[source]: Alias for field number 12

prometheus_pool2: int[source]: Alias for field number 15

prometheus_pool3: int[source]: Alias for field number 18

prometheus_relay1: int[source]: Alias for field number 6

relay1: int[source]: Alias for field number 4

submit_api: int[source]: Alias for field number 2

supervisor: int[source]: Alias for field number 3

webserver: int[source]: Alias for field number 1

class cardano_node_tests.utils.cluster_scripts.LocalScripts(num_pools: int = - 1)[source]

Bases: cardano_node_tests.utils.cluster_scripts.ScriptsTypes

Local cluster scripts (full cardano mode).

copy_scripts_files(destdir: pathlib.Path, scripts_dir: Optional[pathlib.Path] = None) → cardano_node_tests.utils.cluster_scripts.StartupFiles[source]: Make copy of cluster scripts files located in this repository.

get_instance_ports(instance_num: int) → cardano_node_tests.utils.cluster_scripts.InstancePorts[source]: Return ports mapping for given cluster instance.

prepare_scripts_files(destdir: Union[str, pathlib.Path], instance_num: int, start_script: Union[str, pathlib.Path] = '', stop_script: Union[str, pathlib.Path] = '') → cardano_node_tests.utils.cluster_scripts.InstanceFiles[source]: Prepare scripts files for starting and stopping cluster instance.

class cardano_node_tests.utils.cluster_scripts.NodePorts(num, node, ekg, prometheus)[source]

Bases: NamedTuple

ekg: int[source]: Alias for field number 2

node: int[source]: Alias for field number 1

num: int[source]: Alias for field number 0

prometheus: int[source]: Alias for field number 3

class cardano_node_tests.utils.cluster_scripts.ScriptsTypes[source]

Bases: object

Generic cluster scripts.

LOCAL = 'local'[source]

TESTNET = 'testnet'[source]

TESTNET_NOPOOLS = 'testnet_nopools'[source]

copy_scripts_files(destdir: pathlib.Path) → cardano_node_tests.utils.cluster_scripts.StartupFiles[source]: Make copy of cluster scripts files.

get_instance_ports(instance_num: int) → cardano_node_tests.utils.cluster_scripts.InstancePorts[source]: Return ports mapping for given cluster instance.

prepare_scripts_files(destdir: Union[str, pathlib.Path], instance_num: int, start_script: Union[str, pathlib.Path] = '', stop_script: Union[str, pathlib.Path] = '') → cardano_node_tests.utils.cluster_scripts.InstanceFiles[source]: Prepare scripts files for starting and stopping cluster instance.

class cardano_node_tests.utils.cluster_scripts.StartupFiles(start_script, genesis_spec, config_glob)[source]

Bases: NamedTuple

config_glob: str[source]: Alias for field number 2

genesis_spec: pathlib.Path[source]: Alias for field number 1

start_script: pathlib.Path[source]: Alias for field number 0

class cardano_node_tests.utils.cluster_scripts.TestnetNopoolsScripts[source]

Bases: cardano_node_tests.utils.cluster_scripts.TestnetScripts

Testnet cluster scripts (full cardano mode), no pools.

get_instance_ports(instance_num: int) → cardano_node_tests.utils.cluster_scripts.InstancePorts[source]: Return ports mapping for given cluster instance.

class cardano_node_tests.utils.cluster_scripts.TestnetScripts[source]

Bases: cardano_node_tests.utils.cluster_scripts.ScriptsTypes

Testnet cluster scripts (full cardano mode).

BOOTSTRAP_CONF = 'testnet_conf'[source]

TESTNET_GLOBS = ('config*.json', 'genesis-*.json', 'topology-*.json', 'dbsync-config.*', 'submit-api-config.*')[source]

copy_scripts_files(destdir: pathlib.Path) → cardano_node_tests.utils.cluster_scripts.StartupFiles[source]: Make copy of cluster scripts files located in this repository.

get_bootstrap_conf_dir(bootstrap_dir: pathlib.Path) → pathlib.Path[source]

get_instance_ports(instance_num: int) → cardano_node_tests.utils.cluster_scripts.InstancePorts[source]: Return ports mapping for given cluster instance.

prepare_scripts_files(destdir: Union[str, pathlib.Path], instance_num: int, start_script: Union[str, pathlib.Path] = '', stop_script: Union[str, pathlib.Path] = '') → cardano_node_tests.utils.cluster_scripts.InstanceFiles[source]: Prepare scripts files for starting and stopping cluster instance.

cardano_node_tests.utils.clusterlib_utils module

Utilities that extends the functionality of cardano-clusterlib.

class cardano_node_tests.utils.clusterlib_utils.TokenRecord(token, amount, issuers_addrs, token_mint_addr, script)[source]

Bases: NamedTuple

amount: int[source]: Alias for field number 1

issuers_addrs: List[cardano_clusterlib.structs.AddressRecord][source]: Alias for field number 2

script: pathlib.Path[source]: Alias for field number 4

token: str[source]: Alias for field number 0

token_mint_addr: cardano_clusterlib.structs.AddressRecord[source]: Alias for field number 3

class cardano_node_tests.utils.clusterlib_utils.TxMetadata(metadata, aux_data)[source]

Bases: NamedTuple

aux_data: list[source]: Alias for field number 1

metadata: dict[source]: Alias for field number 0

class cardano_node_tests.utils.clusterlib_utils.UpdateProposal(arg, value, name)[source]

Bases: NamedTuple

arg: str[source]: Alias for field number 0

name: str[source]: Alias for field number 2

value: Any[source]: Alias for field number 1

cardano_node_tests.utils.clusterlib_utils.check_pool_data(pool_params: dict, pool_creation_data: cardano_clusterlib.structs.PoolData) → str[source]: Check that actual pool state corresponds with pool creation data.

cardano_node_tests.utils.clusterlib_utils.check_txins_spent(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, txins: List[cardano_clusterlib.structs.UTXOData], wait_blocks: int = 2) → None[source]: Check that txins were spent.

cardano_node_tests.utils.clusterlib_utils.check_updated_params(update_proposals: List[cardano_node_tests.utils.clusterlib_utils.UpdateProposal], protocol_params: dict) → None[source]: Compare update proposals with actual protocol parameters.

cardano_node_tests.utils.clusterlib_utils.cli_has(command: str) → bool[source]

Check if a cardano-cli subcommand or argument is available.

E.g. cli_has(“query leadership-schedule –next”)

cardano_node_tests.utils.clusterlib_utils.create_payment_addr_records(*names: str, cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, stake_vkey_file: Optional[Union[str, pathlib.Path]] = None, destination_dir: Union[str, pathlib.Path] = '.') → List[cardano_clusterlib.structs.AddressRecord][source]: Create new payment address(es).

cardano_node_tests.utils.clusterlib_utils.create_pool_users(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, name_template: str, no_of_addr: int = 1) → List[cardano_clusterlib.structs.PoolUser][source]: Create PoolUsers.

cardano_node_tests.utils.clusterlib_utils.create_reference_utxo(temp_template: str, cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, payment_addr: cardano_clusterlib.structs.AddressRecord, dst_addr: cardano_clusterlib.structs.AddressRecord, script_file: pathlib.Path, amount: int) → Tuple[cardano_clusterlib.structs.UTXOData, cardano_clusterlib.structs.TxRawOutput][source]: Create a reference script UTxO.

cardano_node_tests.utils.clusterlib_utils.create_script_context(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, plutus_version: int, redeemer_file: pathlib.Path, tx_file: Optional[pathlib.Path] = None) → None[source]: Run the create-script-context command (available in plutus-apps).

cardano_node_tests.utils.clusterlib_utils.create_stake_addr_records(*names: str, cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, destination_dir: Union[str, pathlib.Path] = '.') → List[cardano_clusterlib.structs.AddressRecord][source]: Create new stake address(es).

cardano_node_tests.utils.clusterlib_utils.datum_hash_from_txout(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, txout: cardano_clusterlib.structs.TxOut) → str[source]: Return datum hash from clusterlib.TxOut.

cardano_node_tests.utils.clusterlib_utils.deregister_stake_address(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, pool_user: cardano_clusterlib.structs.PoolUser, name_template: str) → Tuple[cardano_clusterlib.structs.TxRawOutput, cardano_clusterlib.structs.TxRawOutput][source]: Deregister stake address.

cardano_node_tests.utils.clusterlib_utils.filtered_ledger_state(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib) → str[source]: Get filtered output of query ledger-state.

cardano_node_tests.utils.clusterlib_utils.fund_from_faucet(*dst_addrs: Union[cardano_clusterlib.structs.AddressRecord, cardano_clusterlib.structs.PoolUser], cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, faucet_data: dict, amount: Union[int, List[int]] = 1000000000, tx_name: Optional[str] = None, destination_dir: Union[str, pathlib.Path] = '.', force: bool = False) → Optional[cardano_clusterlib.structs.TxRawOutput][source]: Send amount from faucet addr to all dst_addrs.

cardano_node_tests.utils.clusterlib_utils.fund_from_genesis(*dst_addrs: str, cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, amount: int = 2000000, tx_name: Optional[str] = None, destination_dir: Union[str, pathlib.Path] = '.') → None[source]: Send amount from genesis addr to all dst_addrs.

cardano_node_tests.utils.clusterlib_utils.gen_byron_addr(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, name_template: str, destination_dir: Union[str, pathlib.Path] = '.') → cardano_clusterlib.structs.AddressRecord[source]: Generate a Byron address and keys.

cardano_node_tests.utils.clusterlib_utils.get_blocks_before(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib) → Dict[str, int][source]: Get blocksBefore section of ledger state with bech32 encoded pool ids.

cardano_node_tests.utils.clusterlib_utils.get_ledger_state(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib) → dict[source]: Return the current ledger state info.

cardano_node_tests.utils.clusterlib_utils.get_pool_state(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, pool_id: str) → cardano_clusterlib.structs.PoolParamsTop[source]: Get pool state using the available command.

cardano_node_tests.utils.clusterlib_utils.get_utxo_ix_offset(utxos: List[cardano_clusterlib.structs.UTXOData], txouts: List[cardano_clusterlib.structs.TxOut]) → int[source]

Get offset of index of the first user-defined txout.

Change txout created by transaction build used to be UTxO with index 0, now it is the last UTxO. This function exists for backwards compatibility with the old behavior.

cardano_node_tests.utils.clusterlib_utils.load_body_metadata(tx_body_file: pathlib.Path) → Any[source]: Load metadata from file containing transaction body.

cardano_node_tests.utils.clusterlib_utils.load_registered_pool_data(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, pool_name: str, pool_id: str) → cardano_clusterlib.structs.PoolData[source]: Load data of existing registered pool.

cardano_node_tests.utils.clusterlib_utils.load_tx_metadata(tx_body_file: pathlib.Path) → cardano_node_tests.utils.clusterlib_utils.TxMetadata[source]: Load transaction metadata from file containing transaction body.

cardano_node_tests.utils.clusterlib_utils.mint_or_burn_sign(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, new_tokens: List[cardano_node_tests.utils.clusterlib_utils.TokenRecord], temp_template: str, sign_incrementally: bool = False) → cardano_clusterlib.structs.TxRawOutput[source]

Mint or burn tokens, depending on the amount value. Sign using skeys.

Positive amount value means minting, negative means burning.

cardano_node_tests.utils.clusterlib_utils.mint_or_burn_witness(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, new_tokens: List[cardano_node_tests.utils.clusterlib_utils.TokenRecord], temp_template: str, invalid_hereafter: Optional[int] = None, invalid_before: Optional[int] = None, use_build_cmd: bool = False, sign_incrementally: bool = False) → cardano_clusterlib.structs.TxRawOutput[source]

Mint or burn tokens, depending on the amount value. Sign using witnesses.

Positive amount value means minting, negative means burning.

cardano_node_tests.utils.clusterlib_utils.new_tokens(*asset_names: str, cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, temp_template: str, token_mint_addr: cardano_clusterlib.structs.AddressRecord, issuer_addr: cardano_clusterlib.structs.AddressRecord, amount: int) → List[cardano_node_tests.utils.clusterlib_utils.TokenRecord][source]: Mint new token, sign using skeys.

cardano_node_tests.utils.clusterlib_utils.register_stake_address(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, pool_user: cardano_clusterlib.structs.PoolUser, name_template: str) → cardano_clusterlib.structs.TxRawOutput[source]: Register stake address.

cardano_node_tests.utils.clusterlib_utils.return_funds_to_faucet(*src_addrs: cardano_clusterlib.structs.AddressRecord, cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, faucet_addr: str, amount: Union[int, List[int]] = - 1, tx_name: Optional[str] = None, destination_dir: Union[str, pathlib.Path] = '.') → None[source]

Send amount from all src_addrs to faucet_addr.

The amount of “-1” means all available funds.

cardano_node_tests.utils.clusterlib_utils.save_ledger_state(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, state_name: str, ledger_state: Optional[dict] = None, destination_dir: Union[str, pathlib.Path] = '.') → pathlib.Path[source]

Save ledger state to file.

Parameters

cluster_obj – An instance of clusterlib.ClusterLib.
state_name – A name of the ledger state (can be epoch number, etc.).
ledger_state – A dict with ledger state to save (optional).
destination_dir – A path to directory for storing the state JSON file (optional).

Returns

A path to the generated state JSON file.

Return type

Path

cardano_node_tests.utils.clusterlib_utils.update_params(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, src_addr_record: cardano_clusterlib.structs.AddressRecord, update_proposals: List[cardano_node_tests.utils.clusterlib_utils.UpdateProposal]) → None[source]: Update params using update proposal.

cardano_node_tests.utils.clusterlib_utils.update_params_build(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, src_addr_record: cardano_clusterlib.structs.AddressRecord, update_proposals: List[cardano_node_tests.utils.clusterlib_utils.UpdateProposal]) → None[source]

Update params using update proposal.

Uses cardano-cli transaction build command for building the transactions.

cardano_node_tests.utils.clusterlib_utils.wait_for_epoch_interval(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, start: int, stop: int, force_epoch: bool = False, check_slot: bool = False) → None[source]

Wait for time interval within an epoch.

Parameters

cluster_obj – An instance of clusterlib.ClusterLib.
start – A start of the interval, in seconds. Negative number for counting from the end of an epoch.
stop – An end of the interval, in seconds. Negative number for counting from the end of an epoch.
force_epoch – A bool indicating whether the interval must be in current epoch (False by default).
check_slot – A bool indicating whether to check if slot number matches time interval after waiting (False by default).

cardano_node_tests.utils.clusterlib_utils.wait_for_rewards(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib) → None[source]: Wait until 4th epoch, if necessary, for first reward distribution.

cardano_node_tests.utils.clusterlib_utils.withdraw_reward_w_build(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, stake_addr_record: cardano_clusterlib.structs.AddressRecord, dst_addr_record: cardano_clusterlib.structs.AddressRecord, tx_name: str, verify: bool = True, destination_dir: Union[str, pathlib.Path] = '.') → cardano_clusterlib.structs.TxRawOutput[source]

Withdraw reward to payment address.

Parameters

cluster_obj – An instance of clusterlib.ClusterLib.
stake_addr_record – An AddressRecord tuple for the stake address with reward.
dst_addr_record – An AddressRecord tuple for the destination payment address.
tx_name – A name of the transaction.
verify – A bool indicating whether to verify that the reward was transferred correctly.
destination_dir – A path to directory for storing artifacts (optional).

cardano_node_tests.utils.configuration module

Cluster and test environment configuration.

cardano_node_tests.utils.dbsync_conn module

Functionality for interacting with db-sync database in postgres.

class cardano_node_tests.utils.dbsync_conn.DBSyncCache[source]

Bases: object

Cache connection to db-sync database for each cluster instance.

conns: Dict[int, Optional[psycopg2.extensions.connection]] = {0: None}[source]

cardano_node_tests.utils.dbsync_conn.close_all() → None[source]

cardano_node_tests.utils.dbsync_conn.conn() → psycopg2.extensions.connection[source]

cardano_node_tests.utils.dbsync_conn.reconn() → psycopg2.extensions.connection[source]

cardano_node_tests.utils.dbsync_queries module

SQL queries to db-sync database.

class cardano_node_tests.utils.dbsync_queries.ADAPotsDBRow(id, slot_no, epoch_no, treasury, reserves, rewards, utxo, deposits, fees, block_id)[source]

Bases: NamedTuple

block_id: int[source]: Alias for field number 9

deposits: decimal.Decimal[source]: Alias for field number 7

epoch_no: int[source]: Alias for field number 2

fees: decimal.Decimal[source]: Alias for field number 8

id: int[source]: Alias for field number 0

reserves: decimal.Decimal[source]: Alias for field number 4

rewards: decimal.Decimal[source]: Alias for field number 5

slot_no: int[source]: Alias for field number 1

treasury: decimal.Decimal[source]: Alias for field number 3

utxo: decimal.Decimal[source]: Alias for field number 6

class cardano_node_tests.utils.dbsync_queries.ADAStashDBRow(id, addr_view, cert_index, amount, tx_id)[source]

Bases: NamedTuple

addr_view: str[source]: Alias for field number 1

amount: decimal.Decimal[source]: Alias for field number 3

cert_index: int[source]: Alias for field number 2

id: int[source]: Alias for field number 0

tx_id: int[source]: Alias for field number 4

class cardano_node_tests.utils.dbsync_queries.BlockDBRow(id, epoch_no, slot_no, epoch_slot_no, block_no, previous_id, tx_count, proto_major, proto_minor, pool_id)[source]

Bases: NamedTuple

block_no: Optional[int][source]: Alias for field number 4

epoch_no: Optional[int][source]: Alias for field number 1

epoch_slot_no: Optional[int][source]: Alias for field number 3

id: int[source]: Alias for field number 0

pool_id: str[source]: Alias for field number 9

previous_id: Optional[int][source]: Alias for field number 5

proto_major: Optional[int][source]: Alias for field number 7

proto_minor: Optional[int][source]: Alias for field number 8

slot_no: Optional[int][source]: Alias for field number 2

tx_count: Optional[int][source]: Alias for field number 6

class cardano_node_tests.utils.dbsync_queries.CollateralTxOutDBRow(tx_out_id, utxo_ix, address, value, tx_hash)[source]

Bases: NamedTuple

address: str[source]: Alias for field number 2

tx_hash: memoryview[source]: Alias for field number 4

tx_out_id: int[source]: Alias for field number 0

utxo_ix: int[source]: Alias for field number 1

value: decimal.Decimal[source]: Alias for field number 3

class cardano_node_tests.utils.dbsync_queries.DatumDBRow(id, datum_hash, tx_id, value, bytes)[source]

Bases: NamedTuple

bytes: memoryview[source]: Alias for field number 4

datum_hash: memoryview[source]: Alias for field number 1

id: int[source]: Alias for field number 0

tx_id: int[source]: Alias for field number 2

value: dict[source]: Alias for field number 3

class cardano_node_tests.utils.dbsync_queries.EpochDBRow(id, out_sum, fees, tx_count, blk_count, epoch_number)[source]

Bases: NamedTuple

blk_count: int[source]: Alias for field number 4

epoch_number: int[source]: Alias for field number 5

fees: int[source]: Alias for field number 2

id: int[source]: Alias for field number 0

out_sum: int[source]: Alias for field number 1

tx_count: int[source]: Alias for field number 3

class cardano_node_tests.utils.dbsync_queries.EpochStakeDBRow(id, hash, view, amount, epoch_number)[source]

Bases: NamedTuple

amount: int[source]: Alias for field number 3

epoch_number: int[source]: Alias for field number 4

hash: memoryview[source]: Alias for field number 1

id: int[source]: Alias for field number 0

view: str[source]: Alias for field number 2

class cardano_node_tests.utils.dbsync_queries.ExtraKeyWitnessDBRow(tx_hash, witness_hash)[source]

Bases: NamedTuple

tx_hash: memoryview[source]: Alias for field number 0

witness_hash: memoryview[source]: Alias for field number 1

class cardano_node_tests.utils.dbsync_queries.MetadataDBRow(id, key, json, bytes, tx_id)[source]

Bases: NamedTuple

bytes: memoryview[source]: Alias for field number 3

id: int[source]: Alias for field number 0

json: Any[source]: Alias for field number 2

key: decimal.Decimal[source]: Alias for field number 1

tx_id: int[source]: Alias for field number 4

class cardano_node_tests.utils.dbsync_queries.ParamProposalDBRow(id, epoch_no, key, min_fee_a, min_fee_b, max_block_size, max_tx_size, max_bh_size, key_deposit, pool_deposit, max_epoch, optimal_pool_count, influence, monetary_expand_rate, treasury_growth_rate, decentralisation, entropy, protocol_major, protocol_minor, min_utxo_value, min_pool_cost, coins_per_utxo_word, cost_model_id, price_mem, price_step, max_tx_ex_mem, max_tx_ex_steps, max_block_ex_mem, max_block_ex_steps, max_val_size, collateral_percent, max_collateral_inputs, registered_tx_id)[source]

Bases: NamedTuple

coins_per_utxo_word: int[source]: Alias for field number 21

collateral_percent: int[source]: Alias for field number 30

cost_model_id: int[source]: Alias for field number 22

decentralisation: float[source]: Alias for field number 15

entropy: memoryview[source]: Alias for field number 16

epoch_no: int[source]: Alias for field number 1

id: int[source]: Alias for field number 0

influence: float[source]: Alias for field number 12

key: memoryview[source]: Alias for field number 2

key_deposit: int[source]: Alias for field number 8

max_bh_size: int[source]: Alias for field number 7

max_block_ex_mem: int[source]: Alias for field number 27

max_block_ex_steps: int[source]: Alias for field number 28

max_block_size: int[source]: Alias for field number 5

max_collateral_inputs: int[source]: Alias for field number 31

max_epoch: int[source]: Alias for field number 10

max_tx_ex_mem: int[source]: Alias for field number 25

max_tx_ex_steps: int[source]: Alias for field number 26

max_tx_size: int[source]: Alias for field number 6

max_val_size: int[source]: Alias for field number 29

min_fee_a: int[source]: Alias for field number 3

min_fee_b: int[source]: Alias for field number 4

min_pool_cost: int[source]: Alias for field number 20

min_utxo_value: int[source]: Alias for field number 19

monetary_expand_rate: float[source]: Alias for field number 13

optimal_pool_count: int[source]: Alias for field number 11

pool_deposit: int[source]: Alias for field number 9

price_mem: float[source]: Alias for field number 23

price_step: float[source]: Alias for field number 24

protocol_major: int[source]: Alias for field number 17

protocol_minor: int[source]: Alias for field number 18

registered_tx_id: int[source]: Alias for field number 32

treasury_growth_rate: float[source]: Alias for field number 14

class cardano_node_tests.utils.dbsync_queries.PoolDataDBRow(id, hash, view, cert_index, vrf_key_hash, pledge, reward_addr, reward_addr_view, active_epoch_no, meta_id, margin, fixed_cost, registered_tx_id, metadata_url, metadata_hash, owner_stake_address_id, owner, ipv4, ipv6, dns_name, port, retire_cert_index, retire_announced_tx_id, retiring_epoch)[source]

Bases: NamedTuple

active_epoch_no: int[source]: Alias for field number 8

cert_index: int[source]: Alias for field number 3

dns_name: str[source]: Alias for field number 19

fixed_cost: int[source]: Alias for field number 11

hash: memoryview[source]: Alias for field number 1

id: int[source]: Alias for field number 0

ipv4: str[source]: Alias for field number 17

ipv6: str[source]: Alias for field number 18

margin: decimal.Decimal[source]: Alias for field number 10

meta_id: int[source]: Alias for field number 9

metadata_hash: memoryview[source]: Alias for field number 14

metadata_url: str[source]: Alias for field number 13

owner: memoryview[source]: Alias for field number 16

owner_stake_address_id: int[source]: Alias for field number 15

pledge: int[source]: Alias for field number 5

port: int[source]: Alias for field number 20

registered_tx_id: int[source]: Alias for field number 12

retire_announced_tx_id: int[source]: Alias for field number 22

retire_cert_index: int[source]: Alias for field number 21

retiring_epoch: int[source]: Alias for field number 23

reward_addr: memoryview[source]: Alias for field number 6

reward_addr_view: str[source]: Alias for field number 7

view: str[source]: Alias for field number 2

vrf_key_hash: memoryview[source]: Alias for field number 4

class cardano_node_tests.utils.dbsync_queries.PoolOfflineDataDBRow(id, ticker_name, hash, json, bytes, pmr_id)[source]

Bases: NamedTuple

bytes: memoryview[source]: Alias for field number 4

hash: memoryview[source]: Alias for field number 2

id: int[source]: Alias for field number 0

json: dict[source]: Alias for field number 3

pmr_id: int[source]: Alias for field number 5

ticker_name: str[source]: Alias for field number 1

class cardano_node_tests.utils.dbsync_queries.PoolOfflineFetchErrorDBRow(id, pmr_id, fetch_error, retry_count)[source]

Bases: NamedTuple

fetch_error: str[source]: Alias for field number 2

id: int[source]: Alias for field number 0

pmr_id: int[source]: Alias for field number 1

retry_count: int[source]: Alias for field number 3

class cardano_node_tests.utils.dbsync_queries.PotTransferDBRow(id, cert_index, treasury, reserves, tx_id)[source]

Bases: NamedTuple

cert_index: int[source]: Alias for field number 1

id: int[source]: Alias for field number 0

reserves: decimal.Decimal[source]: Alias for field number 3

treasury: decimal.Decimal[source]: Alias for field number 2

tx_id: int[source]: Alias for field number 4

class cardano_node_tests.utils.dbsync_queries.RedeemerDBRow(id, tx_id, unit_mem, unit_steps, fee, purpose, script_hash, value)[source]

Bases: NamedTuple

fee: int[source]: Alias for field number 4

id: int[source]: Alias for field number 0

purpose: str[source]: Alias for field number 5

script_hash: memoryview[source]: Alias for field number 6

tx_id: int[source]: Alias for field number 1

unit_mem: int[source]: Alias for field number 2

unit_steps: int[source]: Alias for field number 3

value: dict[source]: Alias for field number 7

class cardano_node_tests.utils.dbsync_queries.RewardDBRow(address, type, amount, earned_epoch, spendable_epoch, pool_id)[source]

Bases: NamedTuple

address: str[source]: Alias for field number 0

amount: decimal.Decimal[source]: Alias for field number 2

earned_epoch: int[source]: Alias for field number 3

pool_id: str[source]: Alias for field number 5

spendable_epoch: int[source]: Alias for field number 4

type: str[source]: Alias for field number 1

class cardano_node_tests.utils.dbsync_queries.SchemaVersion[source]

Bases: object

Query and cache db-sync schema version.

classmethod stages() → cardano_node_tests.utils.dbsync_queries.SchemaVersionStages[source]

class cardano_node_tests.utils.dbsync_queries.SchemaVersionStages(one, two, three)[source]

Bases: NamedTuple

one: int[source]: Alias for field number 0

three: int[source]: Alias for field number 2

two: int[source]: Alias for field number 1

class cardano_node_tests.utils.dbsync_queries.ScriptDBRow(id, tx_id, hash, type, serialised_size)[source]

Bases: NamedTuple

hash: memoryview[source]: Alias for field number 2

id: int[source]: Alias for field number 0

serialised_size: Optional[int][source]: Alias for field number 4

tx_id: int[source]: Alias for field number 1

type: str[source]: Alias for field number 3

class cardano_node_tests.utils.dbsync_queries.StakeAddrDBRow(id, view, tx_id)[source]

Bases: NamedTuple

id: int[source]: Alias for field number 0

tx_id: int[source]: Alias for field number 2

view: str[source]: Alias for field number 1

class cardano_node_tests.utils.dbsync_queries.StakeDelegDBRow(tx_id, active_epoch_no, pool_id, address)[source]

Bases: NamedTuple

active_epoch_no: Optional[int][source]: Alias for field number 1

address: Optional[str][source]: Alias for field number 3

pool_id: Optional[str][source]: Alias for field number 2

tx_id: int[source]: Alias for field number 0

class cardano_node_tests.utils.dbsync_queries.TxDBRow(tx_id, tx_hash, block_id, block_index, out_sum, fee, deposit, size, invalid_before, invalid_hereafter, tx_out_id, tx_out_tx_id, utxo_ix, tx_out_addr, tx_out_addr_has_script, tx_out_value, tx_out_data_hash, tx_out_inline_datum_hash, tx_out_reference_script_hash, metadata_count, reserve_count, treasury_count, pot_transfer_count, stake_reg_count, stake_dereg_count, stake_deleg_count, withdrawal_count, collateral_count, reference_input_count, collateral_out_count, script_count, redeemer_count, extra_key_witness_count, ma_tx_out_id, ma_tx_out_policy, ma_tx_out_name, ma_tx_out_quantity, ma_tx_mint_id, ma_tx_mint_policy, ma_tx_mint_name, ma_tx_mint_quantity)[source]

Bases: NamedTuple

block_id: int[source]: Alias for field number 2

block_index: int[source]: Alias for field number 3

collateral_count: int[source]: Alias for field number 27

collateral_out_count: int[source]: Alias for field number 29

deposit: int[source]: Alias for field number 6

extra_key_witness_count: int[source]: Alias for field number 32

fee: decimal.Decimal[source]: Alias for field number 5

invalid_before: Optional[decimal.Decimal][source]: Alias for field number 8

invalid_hereafter: Optional[decimal.Decimal][source]: Alias for field number 9

ma_tx_mint_id: Optional[int][source]: Alias for field number 37

ma_tx_mint_name: Optional[memoryview][source]: Alias for field number 39

ma_tx_mint_policy: Optional[memoryview][source]: Alias for field number 38

ma_tx_mint_quantity: Optional[decimal.Decimal][source]: Alias for field number 40

ma_tx_out_id: Optional[int][source]: Alias for field number 33

ma_tx_out_name: Optional[memoryview][source]: Alias for field number 35

ma_tx_out_policy: Optional[memoryview][source]: Alias for field number 34

ma_tx_out_quantity: Optional[decimal.Decimal][source]: Alias for field number 36

metadata_count: int[source]: Alias for field number 19

out_sum: decimal.Decimal[source]: Alias for field number 4

pot_transfer_count: int[source]: Alias for field number 22

redeemer_count: int[source]: Alias for field number 31

reference_input_count: int[source]: Alias for field number 28

reserve_count: int[source]: Alias for field number 20

script_count: int[source]: Alias for field number 30

size: int[source]: Alias for field number 7

stake_deleg_count: int[source]: Alias for field number 25

stake_dereg_count: int[source]: Alias for field number 24

stake_reg_count: int[source]: Alias for field number 23

treasury_count: int[source]: Alias for field number 21

tx_hash: memoryview[source]: Alias for field number 1

tx_id: int[source]: Alias for field number 0

tx_out_addr: str[source]: Alias for field number 13

tx_out_addr_has_script: bool[source]: Alias for field number 14

tx_out_data_hash: Optional[memoryview][source]: Alias for field number 16

tx_out_id: int[source]: Alias for field number 10

tx_out_inline_datum_hash: Optional[memoryview][source]: Alias for field number 17

tx_out_reference_script_hash: Optional[memoryview][source]: Alias for field number 18

tx_out_tx_id: int[source]: Alias for field number 11

tx_out_value: decimal.Decimal[source]: Alias for field number 15

utxo_ix: int[source]: Alias for field number 12

withdrawal_count: int[source]: Alias for field number 26

class cardano_node_tests.utils.dbsync_queries.TxInDBRow(tx_out_id, utxo_ix, address, value, tx_hash, ma_tx_out_id, ma_tx_out_policy, ma_tx_out_name, ma_tx_out_quantity)[source]

Bases: NamedTuple

address: str[source]: Alias for field number 2

ma_tx_out_id: Optional[int][source]: Alias for field number 5

ma_tx_out_name: Optional[memoryview][source]: Alias for field number 7

ma_tx_out_policy: Optional[memoryview][source]: Alias for field number 6

ma_tx_out_quantity: Optional[decimal.Decimal][source]: Alias for field number 8

tx_hash: memoryview[source]: Alias for field number 4

tx_out_id: int[source]: Alias for field number 0

utxo_ix: int[source]: Alias for field number 1

value: decimal.Decimal[source]: Alias for field number 3

class cardano_node_tests.utils.dbsync_queries.TxInNoMADBRow(tx_out_id, utxo_ix, address, value, tx_hash)[source]

Bases: NamedTuple

address: str[source]: Alias for field number 2

tx_hash: memoryview[source]: Alias for field number 4

tx_out_id: int[source]: Alias for field number 0

utxo_ix: int[source]: Alias for field number 1

value: decimal.Decimal[source]: Alias for field number 3

class cardano_node_tests.utils.dbsync_queries.UTxODBRow(tx_hash, utxo_ix, payment_address, stake_address, has_script, value, data_hash)[source]

Bases: NamedTuple

data_hash: Optional[memoryview][source]: Alias for field number 6

has_script: bool[source]: Alias for field number 4

payment_address: str[source]: Alias for field number 2

stake_address: str[source]: Alias for field number 3

tx_hash: memoryview[source]: Alias for field number 0

utxo_ix: int[source]: Alias for field number 1

value: int[source]: Alias for field number 5

class cardano_node_tests.utils.dbsync_queries.WithdrawalDBRow(tx_id, address, amount)[source]

Bases: NamedTuple

address: str[source]: Alias for field number 1

amount: int[source]: Alias for field number 2

tx_id: int[source]: Alias for field number 0

cardano_node_tests.utils.dbsync_queries.execute(query: str, vars: Sequence = ()) → Iterator[psycopg2.extensions.cursor][source]

cardano_node_tests.utils.dbsync_queries.query_ada_pots(epoch_from: int = 0, epoch_to: int = 99999999) → Generator[cardano_node_tests.utils.dbsync_queries.ADAPotsDBRow, None, None][source]: Query ADA pots record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_address_reward(address: str, epoch_from: int = 0, epoch_to: int = 99999999) → Generator[cardano_node_tests.utils.dbsync_queries.RewardDBRow, None, None][source]: Query reward records for stake address in db-sync.

cardano_node_tests.utils.dbsync_queries.query_blocks(pool_id_bech32: str = '', epoch_from: int = 0, epoch_to: int = 99999999) → Generator[cardano_node_tests.utils.dbsync_queries.BlockDBRow, None, None][source]: Query block records in db-sync.

cardano_node_tests.utils.dbsync_queries.query_collateral_tx_ins(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.TxInNoMADBRow, None, None][source]: Query transaction collateral txins in db-sync.

cardano_node_tests.utils.dbsync_queries.query_collateral_tx_outs(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.CollateralTxOutDBRow, None, None][source]: Query transaction collateral txouts in db-sync.

cardano_node_tests.utils.dbsync_queries.query_cost_model() → Dict[str, Dict[str, Any]][source]: Query last cost-model record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_datum(datum_hash: str) → Generator[cardano_node_tests.utils.dbsync_queries.DatumDBRow, None, None][source]: Query datum record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_epoch(epoch_from: int = 0, epoch_to: int = 99999999) → Generator[cardano_node_tests.utils.dbsync_queries.EpochDBRow, None, None][source]: Query epoch records in db-sync.

cardano_node_tests.utils.dbsync_queries.query_epoch_stake(pool_id_bech32: str, epoch_number: int) → Generator[cardano_node_tests.utils.dbsync_queries.EpochStakeDBRow, None, None][source]: Query epoch stake record for a pool in db-sync.

cardano_node_tests.utils.dbsync_queries.query_extra_key_witness(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.ExtraKeyWitnessDBRow, None, None][source]: Query extra key witness records in db-sync.

cardano_node_tests.utils.dbsync_queries.query_param_proposal() → cardano_node_tests.utils.dbsync_queries.ParamProposalDBRow[source]: Query last param proposal record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_pool_data(pool_id_bech32: str) → Generator[cardano_node_tests.utils.dbsync_queries.PoolDataDBRow, None, None][source]: Query pool data record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_pool_offline_data(pool_id_bech32: str) → Generator[cardano_node_tests.utils.dbsync_queries.PoolOfflineDataDBRow, None, None][source]: Query PoolOfflineData record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_pool_offline_fetch_error(pool_id_bech32: str) → Generator[cardano_node_tests.utils.dbsync_queries.PoolOfflineFetchErrorDBRow, None, None][source]: Query PoolOfflineFetchError record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_redeemers(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.RedeemerDBRow, None, None][source]: Query transaction redeemers in db-sync.

cardano_node_tests.utils.dbsync_queries.query_reference_tx_ins(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.TxInNoMADBRow, None, None][source]: Query transaction reference txins in db-sync.

cardano_node_tests.utils.dbsync_queries.query_scripts(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.ScriptDBRow, None, None][source]: Query transaction scripts in db-sync.

cardano_node_tests.utils.dbsync_queries.query_table_names() → List[str][source]: Query table names in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.TxDBRow, None, None][source]: Query a transaction in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_ins(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.TxInDBRow, None, None][source]: Query transaction txins in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_metadata(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.MetadataDBRow, None, None][source]: Query transaction metadata in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_pot_transfers(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.PotTransferDBRow, None, None][source]: Query transaction MIR certificate records in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_reserve(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.ADAStashDBRow, None, None][source]: Query transaction reserve record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_stake_deleg(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.StakeDelegDBRow, None, None][source]: Query stake registration record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_stake_dereg(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.StakeAddrDBRow, None, None][source]: Query stake deregistration record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_stake_reg(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.StakeAddrDBRow, None, None][source]: Query stake registration record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_treasury(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.ADAStashDBRow, None, None][source]: Query transaction treasury record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_withdrawal(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.WithdrawalDBRow, None, None][source]: Query reward withdrawal record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_utxo(address: str) → Generator[cardano_node_tests.utils.dbsync_queries.UTxODBRow, None, None][source]: Query UTxOs for payment address in db-sync.

cardano_node_tests.utils.dbsync_utils module

Functionality for interacting with db-sync.

class cardano_node_tests.utils.dbsync_utils.ADAStashRecord(address, cert_index, amount)[source]

Bases: NamedTuple

address: str[source]: Alias for field number 0

amount: int[source]: Alias for field number 2

cert_index: int[source]: Alias for field number 1

class cardano_node_tests.utils.dbsync_utils.DelegationRecord(address, pool_id, active_epoch_no)[source]

Bases: NamedTuple

active_epoch_no: int[source]: Alias for field number 2

address: str[source]: Alias for field number 0

pool_id: str[source]: Alias for field number 1

class cardano_node_tests.utils.dbsync_utils.ExtraKeyWitnessRecord(tx_hash, witness_hash)[source]

Bases: NamedTuple

tx_hash: str[source]: Alias for field number 0

witness_hash: str[source]: Alias for field number 1

class cardano_node_tests.utils.dbsync_utils.GetUTxORecord(utxo_hash, utxo_ix, has_script, amount, data_hash)[source]

Bases: NamedTuple

amount: int[source]: Alias for field number 3

data_hash: str[source]: Alias for field number 4

has_script: bool[source]: Alias for field number 2

utxo_hash: str[source]: Alias for field number 0

utxo_ix: int[source]: Alias for field number 1

class cardano_node_tests.utils.dbsync_utils.MetadataRecord(key, json, bytes)[source]

Bases: NamedTuple

bytes: memoryview[source]: Alias for field number 2

json: Any[source]: Alias for field number 1

key: int[source]: Alias for field number 0

class cardano_node_tests.utils.dbsync_utils.PaymentAddrRecord(payment_address, stake_address, amount_sum, utxos)[source]

Bases: NamedTuple

amount_sum: int[source]: Alias for field number 2

payment_address: str[source]: Alias for field number 0

stake_address: Optional[str][source]: Alias for field number 1

utxos: List[cardano_node_tests.utils.dbsync_utils.GetUTxORecord][source]: Alias for field number 3

class cardano_node_tests.utils.dbsync_utils.PoolDataRecord(id, hash, view, cert_index, vrf_key_hash, pledge, reward_addr, active_epoch_no, meta_id, margin, fixed_cost, registered_tx_id, metadata_url, metadata_hash, owners, relays, retire_cert_index, retire_announced_tx_id, retiring_epoch)[source]

Bases: NamedTuple

active_epoch_no: int[source]: Alias for field number 7

cert_index: int[source]: Alias for field number 3

fixed_cost: int[source]: Alias for field number 10

hash: str[source]: Alias for field number 1

id: int[source]: Alias for field number 0

margin: float[source]: Alias for field number 9

meta_id: int[source]: Alias for field number 8

metadata_hash: str[source]: Alias for field number 13

metadata_url: str[source]: Alias for field number 12

owners: List[str][source]: Alias for field number 14

pledge: int[source]: Alias for field number 5

registered_tx_id: int[source]: Alias for field number 11

relays: List[Dict[str, Dict[str, Any]]][source]: Alias for field number 15

retire_announced_tx_id: int[source]: Alias for field number 17

retire_cert_index: int[source]: Alias for field number 16

retiring_epoch: int[source]: Alias for field number 18

reward_addr: str[source]: Alias for field number 6

view: str[source]: Alias for field number 2

vrf_key_hash: str[source]: Alias for field number 4

class cardano_node_tests.utils.dbsync_utils.PotTransferRecord(treasury, reserves)[source]

Bases: NamedTuple

reserves: int[source]: Alias for field number 1

treasury: int[source]: Alias for field number 0

class cardano_node_tests.utils.dbsync_utils.RedeemerRecord(unit_mem, unit_steps, fee, purpose, script_hash, value)[source]

Bases: NamedTuple

fee: int[source]: Alias for field number 2

purpose: str[source]: Alias for field number 3

script_hash: str[source]: Alias for field number 4

unit_mem: int[source]: Alias for field number 0

unit_steps: int[source]: Alias for field number 1

value: dict[source]: Alias for field number 5

class cardano_node_tests.utils.dbsync_utils.RewardEpochRecord(amount, earned_epoch, spendable_epoch, type, pool_id)[source]

Bases: NamedTuple

amount: int[source]: Alias for field number 0

earned_epoch: int[source]: Alias for field number 1

pool_id: str[source]: Alias for field number 4

spendable_epoch: int[source]: Alias for field number 2

type: str[source]: Alias for field number 3

class cardano_node_tests.utils.dbsync_utils.RewardRecord(address, rewards, reward_sum)[source]

Bases: NamedTuple

address: str[source]: Alias for field number 0

reward_sum: int[source]: Alias for field number 2

rewards: List[cardano_node_tests.utils.dbsync_utils.RewardEpochRecord][source]: Alias for field number 1

class cardano_node_tests.utils.dbsync_utils.ScriptRecord(hash, type, serialised_size)[source]

Bases: NamedTuple

hash: str[source]: Alias for field number 0

serialised_size: int[source]: Alias for field number 2

type: str[source]: Alias for field number 1

class cardano_node_tests.utils.dbsync_utils.TxPrelimRecord(utxo_out, ma_utxo_out, mint_utxo_out, last_row)[source]

Bases: NamedTuple

last_row: cardano_node_tests.utils.dbsync_queries.TxDBRow[source]: Alias for field number 3

ma_utxo_out: List[cardano_node_tests.utils.dbsync_utils.UTxORecord][source]: Alias for field number 1

mint_utxo_out: List[cardano_node_tests.utils.dbsync_utils.UTxORecord][source]: Alias for field number 2

utxo_out: List[cardano_node_tests.utils.dbsync_utils.UTxORecord][source]: Alias for field number 0

class cardano_node_tests.utils.dbsync_utils.TxRecord(tx_id, tx_hash, block_id, block_index, out_sum, fee, deposit, size, invalid_before, invalid_hereafter, txins, txouts, mint, collaterals, collateral_outputs, reference_inputs, scripts, redeemers, metadata, reserve, treasury, pot_transfers, stake_registration, stake_deregistration, stake_delegation, withdrawals, extra_key_witness)[source]

Bases: NamedTuple

block_id: int[source]: Alias for field number 2

block_index: int[source]: Alias for field number 3

collateral_outputs: List[cardano_clusterlib.structs.UTXOData][source]: Alias for field number 14

collaterals: List[cardano_clusterlib.structs.UTXOData][source]: Alias for field number 13

deposit: int[source]: Alias for field number 6

extra_key_witness: List[cardano_node_tests.utils.dbsync_utils.ExtraKeyWitnessRecord][source]: Alias for field number 26

fee: int[source]: Alias for field number 5

invalid_before: Optional[int][source]: Alias for field number 8

invalid_hereafter: Optional[int][source]: Alias for field number 9

metadata: List[cardano_node_tests.utils.dbsync_utils.MetadataRecord][source]: Alias for field number 18

mint: List[cardano_node_tests.utils.dbsync_utils.UTxORecord][source]: Alias for field number 12

out_sum: int[source]: Alias for field number 4

pot_transfers: List[cardano_node_tests.utils.dbsync_utils.PotTransferRecord][source]: Alias for field number 21

redeemers: List[cardano_node_tests.utils.dbsync_utils.RedeemerRecord][source]: Alias for field number 17

reference_inputs: List[cardano_clusterlib.structs.UTXOData][source]: Alias for field number 15

reserve: List[cardano_node_tests.utils.dbsync_utils.ADAStashRecord][source]: Alias for field number 19

scripts: List[cardano_node_tests.utils.dbsync_utils.ScriptRecord][source]: Alias for field number 16

size: int[source]: Alias for field number 7

stake_delegation: List[cardano_node_tests.utils.dbsync_utils.DelegationRecord][source]: Alias for field number 24

stake_deregistration: List[str][source]: Alias for field number 23

stake_registration: List[str][source]: Alias for field number 22

treasury: List[cardano_node_tests.utils.dbsync_utils.ADAStashRecord][source]: Alias for field number 20

tx_hash: str[source]: Alias for field number 1

tx_id: int[source]: Alias for field number 0

txins: List[cardano_clusterlib.structs.UTXOData][source]: Alias for field number 10

txouts: List[cardano_node_tests.utils.dbsync_utils.UTxORecord][source]: Alias for field number 11

withdrawals: List[cardano_clusterlib.structs.TxOut][source]: Alias for field number 25

class cardano_node_tests.utils.dbsync_utils.UTxORecord(utxo_hash, utxo_ix, amount, address, coin, decoded_coin, datum_hash, inline_datum_hash, reference_script_hash)[source]

Bases: NamedTuple

address: str[source]: Alias for field number 3

amount: int[source]: Alias for field number 2

coin: str[source]: Alias for field number 4

datum_hash: str[source]: Alias for field number 6

decoded_coin: str[source]: Alias for field number 5

inline_datum_hash: str[source]: Alias for field number 7

reference_script_hash: str[source]: Alias for field number 8

utxo_hash: str[source]: Alias for field number 0

utxo_ix: int[source]: Alias for field number 1

cardano_node_tests.utils.dbsync_utils.check_address_reward(address: str, epoch_from: int = 0, epoch_to: int = 99999999) → cardano_node_tests.utils.dbsync_utils.RewardRecord[source]

Check reward data for stake address in db-sync.

The epoch_from and epoch_to are epochs where the reward can be spent.

cardano_node_tests.utils.dbsync_utils.check_param_proposal(protocol_params: dict) → Optional[cardano_node_tests.utils.dbsync_queries.ParamProposalDBRow][source]: Check expected values in the param_proposal table in db-sync.

cardano_node_tests.utils.dbsync_utils.check_plutus_cost(redeemer_record: cardano_node_tests.utils.dbsync_utils.RedeemerRecord, cost_record: Dict[str, Any]) → None[source]: Compare cost of Plutus script with data from db-sync.

cardano_node_tests.utils.dbsync_utils.check_plutus_costs(redeemer_records: List[cardano_node_tests.utils.dbsync_utils.RedeemerRecord], cost_records: List[Dict[str, Any]]) → None[source]: Compare cost of multiple Plutus scripts with data from db-sync.

cardano_node_tests.utils.dbsync_utils.check_pool_data(ledger_pool_data: dict, pool_id: str) → Optional[cardano_node_tests.utils.dbsync_utils.PoolDataRecord][source]: Check comparison for pool data between ledger and db-sync.

cardano_node_tests.utils.dbsync_utils.check_pool_deregistration(pool_id: str, retiring_epoch: int) → Optional[cardano_node_tests.utils.dbsync_utils.PoolDataRecord][source]: Check pool retirement in db-sync.

cardano_node_tests.utils.dbsync_utils.check_pool_offline_data(ledger_pool_data: dict, pool_id: str) → cardano_node_tests.utils.dbsync_queries.PoolOfflineDataDBRow[source]: Check comparison for pool offline data between ledger and db-sync.

cardano_node_tests.utils.dbsync_utils.check_pool_offline_fetch_error(ledger_pool_data: dict, pool_id: str) → cardano_node_tests.utils.dbsync_queries.PoolOfflineFetchErrorDBRow[source]: Check expected error on PoolOfflineFetchError.

cardano_node_tests.utils.dbsync_utils.check_tx(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_raw_output: cardano_clusterlib.structs.TxRawOutput, retry_num: int = 3) → Optional[cardano_node_tests.utils.dbsync_utils.TxRecord][source]: Check a transaction in db-sync.

cardano_node_tests.utils.dbsync_utils.check_tx_phase_2_failure(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_raw_output: cardano_clusterlib.structs.TxRawOutput, collateral_charged: int, retry_num: int = 3) → Optional[cardano_node_tests.utils.dbsync_utils.TxRecord][source]: Check a transaction in db-sync when a phase 2 failure happens.

cardano_node_tests.utils.dbsync_utils.get_address_reward(address: str, epoch_from: int = 0, epoch_to: int = 99999999) → cardano_node_tests.utils.dbsync_utils.RewardRecord[source]

Get reward data for stake address from db-sync.

The epoch_from and epoch_to are epochs where the reward can be spent.

cardano_node_tests.utils.dbsync_utils.get_pool_data(pool_id_bech32: str) → Optional[cardano_node_tests.utils.dbsync_utils.PoolDataRecord][source]: Get pool data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_prelim_tx_record(txhash: str) → cardano_node_tests.utils.dbsync_utils.TxPrelimRecord[source]: Get first batch of transaction data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_tx_record(txhash: str) → cardano_node_tests.utils.dbsync_utils.TxRecord[source]

Get transaction data from db-sync.

Compile data from multiple SQL queries to get as much information about the TX as possible.

cardano_node_tests.utils.dbsync_utils.get_tx_record_retry(txhash: str, retry_num: int = 3) → cardano_node_tests.utils.dbsync_utils.TxRecord[source]

Retry get_tx_record when data is anticipated and are not available yet.

Under load it might be necessary to wait a bit and retry the query.

cardano_node_tests.utils.dbsync_utils.get_txins(txhash: str) → List[cardano_clusterlib.structs.UTXOData][source]: Get txins of a transaction from db-sync.

cardano_node_tests.utils.dbsync_utils.get_utxo(address: str) → cardano_node_tests.utils.dbsync_utils.PaymentAddrRecord[source]: Return UTxO info for payment address from db-sync.

cardano_node_tests.utils.dbsync_utils.retry_query(query_func: Callable, timeout: int = 20) → Any[source]

Wait a bit and retry a query until response is returned.

A generic function that can be used by any query/check that raises AssertionError with NO_REPONSE_STR until the expected data is returned.

cardano_node_tests.utils.dbsync_utils.utxodata2txout(utxodata: Union[cardano_node_tests.utils.dbsync_utils.UTxORecord, cardano_clusterlib.structs.UTXOData]) → cardano_clusterlib.structs.TxOut[source]: Convert UTxORecord or UTxOData to clusterlib.TxOut.

cardano_node_tests.utils.helpers module

cardano_node_tests.utils.helpers.callonce(func: cardano_node_tests.utils.helpers.TCallable) → cardano_node_tests.utils.helpers.TCallable[source]: Call a function and cache its return value.

cardano_node_tests.utils.helpers.change_cwd(dir_path: Union[str, pathlib.Path]) → Iterator[Union[str, pathlib.Path]][source]: Change and restore CWD - context manager.

cardano_node_tests.utils.helpers.check_dir_arg(dir_path: str) → Optional[pathlib.Path][source]: Check that the dir passed as argparse parameter is a valid existing dir.

cardano_node_tests.utils.helpers.check_file_arg(file_path: str) → Optional[pathlib.Path][source]: Check that the file passed as argparse parameter is a valid existing file.

cardano_node_tests.utils.helpers.checksum(filename: Union[str, pathlib.Path], blocksize: int = 65536) → str[source]: Return file checksum.

cardano_node_tests.utils.helpers.decode_bech32(bech32: str) → str[source]: Convert from bech32 string.

cardano_node_tests.utils.helpers.encode_bech32(prefix: str, data: str) → str[source]: Convert to bech32 string.

cardano_node_tests.utils.helpers.environ(env: dict) → Iterator[None][source]: Temporarily set environment variables and restore previous environment afterwards.

cardano_node_tests.utils.helpers.get_cmd_path(cmd: str) → pathlib.Path[source]: Return file path of a command.

cardano_node_tests.utils.helpers.get_current_commit() → str[source]

cardano_node_tests.utils.helpers.get_eof_offset(infile: pathlib.Path) → int[source]: Return position of the current end of the file.

cardano_node_tests.utils.helpers.get_rand_str(length: int = 8) → str[source]: Return random string.

cardano_node_tests.utils.helpers.get_timestamped_rand_str(rand_str_length: int = 4) → str[source]

Return random string prefixed with timestamp.

>>> len(get_timestamped_rand_str()) == len("200801_002401314_cinf")
True

cardano_node_tests.utils.helpers.get_vcs_link() → str[source]: Return link to the current line in GitHub.

cardano_node_tests.utils.helpers.ignore_interrupt() → Iterator[None][source]: Ignore the KeyboardInterrupt signal.

cardano_node_tests.utils.helpers.is_in_interval(num1: float, num2: float, frac: float = 0.1) → bool[source]: Check that the num1 is in the interval defined by num2 and its fraction.

cardano_node_tests.utils.helpers.is_port_open(host: str, port: int) → bool[source]: Check if port is open.

cardano_node_tests.utils.helpers.prepend_flag(flag: str, contents: Iterable) → List[str][source]

Prepend flag to every item of the sequence.

Parameters

flag – A flag to prepend to every item of the contents.
contents – A list (iterable) of content to be prepended.

Returns

A list of flag followed by content, see below.

Return type

List[str]

>>> prepend_flag(None, "--foo", [1, 2, 3])
['--foo', '1', '--foo', '2', '--foo', '3']

cardano_node_tests.utils.helpers.replace_str_in_file(infile: pathlib.Path, outfile: pathlib.Path, orig_str: str, new_str: str) → None[source]: Replace a string in file with another string.

cardano_node_tests.utils.helpers.run_command(command: Union[str, list], workdir: Union[str, pathlib.Path] = '', ignore_fail: bool = False, shell: bool = False) → bytes[source]: Run command.

cardano_node_tests.utils.helpers.run_in_bash(command: str, workdir: Union[str, pathlib.Path] = '') → bytes[source]: Run command(s) in bash.

cardano_node_tests.utils.helpers.tool_has(command: str) → bool[source]

Check if a tool has a subcommand or argument available.

E.g. tool_has_arg(“create-script-context –plutus-v1”)

cardano_node_tests.utils.helpers.wait_for(func: Callable, delay: int = 5, num_sec: int = 180, message: str = '', silent: bool = False) → Any[source]: Wait for success of func for num_sec.

cardano_node_tests.utils.helpers.write_json(location: Union[str, pathlib.Path], content: dict) → Union[str, pathlib.Path][source]: Write dictionary content to JSON file.

cardano_node_tests.utils.locking module

cardano_node_tests.utils.logfiles module

class cardano_node_tests.utils.logfiles.RotableLog(logfile, seek, timestamp)[source]

Bases: NamedTuple

logfile: pathlib.Path[source]: Alias for field number 0

seek: int[source]: Alias for field number 1

timestamp: float[source]: Alias for field number 2

cardano_node_tests.utils.logfiles.add_ignore_rule(files_glob: str, regex: str, ignore_file_id: str) → None[source]: Add ignore rule for expected errors.

cardano_node_tests.utils.logfiles.clean_ignore_rules(ignore_file_id: str) → None[source]

Cleanup relevant ignore rules file.

Delete ignore file identified by ignore_file_id when it is no longer valid.

cardano_node_tests.utils.logfiles.expect_errors(regex_pairs: List[Tuple[str, str]], ignore_file_id: str) → Iterator[None][source]

Make sure the expected errors are present in logs.

Parameters

regex_pairs – [(glob, regex)] - A list of regexes that need to be present in files described by the glob.
ignore_file_id – The id of a ignore file the expected error will be added to.

cardano_node_tests.utils.logfiles.get_logfiles_errors() → str[source]: Get errors found in cluster artifacts.

cardano_node_tests.utils.logfiles.search_cluster_logs() → List[Tuple[pathlib.Path, str]][source]: Search cluster logs for errors.

cardano_node_tests.utils.model_ekg module

class cardano_node_tests.utils.model_ekg.Adopted(*, int: cardano_node_tests.utils.model_ekg.Int7)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int7[source]

class cardano_node_tests.utils.model_ekg.BlockNum(*, int: cardano_node_tests.utils.model_ekg.Int1)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int1[source]

class cardano_node_tests.utils.model_ekg.BlocksForgedNum(*, int: cardano_node_tests.utils.model_ekg.Int2)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int2[source]

class cardano_node_tests.utils.model_ekg.BytesAllocated(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.BytesCopied(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Cardano(*, node: cardano_node_tests.utils.model_ekg.Node)[source]

Bases: pydantic.main.BaseModel

node: cardano_node_tests.utils.model_ekg.Node[source]

class cardano_node_tests.utils.model_ekg.Counter(*, int: cardano_node_tests.utils.model_ekg.Int20)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int20[source]

class cardano_node_tests.utils.model_ekg.CpuMs(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Cputicks(*, int: cardano_node_tests.utils.model_ekg.Int29)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int29[source]

class cardano_node_tests.utils.model_ekg.CumulativeBytesUsed(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.CurrentBytesSlop(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.CurrentBytesUsed(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.CurrentKESPeriod(*, int: cardano_node_tests.utils.model_ekg.Int9)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int9[source]

class cardano_node_tests.utils.model_ekg.DelegMapSize(*, int: cardano_node_tests.utils.model_ekg.Int18)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int18[source]

class cardano_node_tests.utils.model_ekg.Density(*, real: cardano_node_tests.utils.model_ekg.Real)[source]

Bases: pydantic.main.BaseModel

real: cardano_node_tests.utils.model_ekg.Real[source]

class cardano_node_tests.utils.model_ekg.Ekg(*, server_timestamp_ms: cardano_node_tests.utils.model_ekg.ServerTimestampMs)[source]

Bases: pydantic.main.BaseModel

server_timestamp_ms: cardano_node_tests.utils.model_ekg.ServerTimestampMs[source]

class cardano_node_tests.utils.model_ekg.Epoch(*, int: cardano_node_tests.utils.model_ekg.Int3)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int3[source]

class cardano_node_tests.utils.model_ekg.Forge(*, forged: cardano_node_tests.utils.model_ekg.Forged, adopted: cardano_node_tests.utils.model_ekg.Adopted, **extra_data: Any)[source]

Bases: pydantic.main.BaseModel

adopted: cardano_node_tests.utils.model_ekg.Adopted[source]

forge_about_to_lead: cardano_node_tests.utils.model_ekg.ForgeAboutToLead[source]

forged: cardano_node_tests.utils.model_ekg.Forged[source]

node_is_leader: cardano_node_tests.utils.model_ekg.NodeIsLeader[source]

node_not_leader: cardano_node_tests.utils.model_ekg.NodeNotLeader[source]

class cardano_node_tests.utils.model_ekg.ForgeAboutToLead(*, int: cardano_node_tests.utils.model_ekg.Int8)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int8[source]

class cardano_node_tests.utils.model_ekg.Forged(*, int: cardano_node_tests.utils.model_ekg.Int6)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int6[source]

Bases: pydantic.main.BaseModel

bytes_allocated: cardano_node_tests.utils.model_ekg.BytesAllocated[source]

bytes_copied: cardano_node_tests.utils.model_ekg.BytesCopied[source]

cpu_ms: cardano_node_tests.utils.model_ekg.CpuMs[source]

cumulative_bytes_used: cardano_node_tests.utils.model_ekg.CumulativeBytesUsed[source]

current_bytes_slop: cardano_node_tests.utils.model_ekg.CurrentBytesSlop[source]

current_bytes_used: cardano_node_tests.utils.model_ekg.CurrentBytesUsed[source]

gc_cpu_ms: cardano_node_tests.utils.model_ekg.GcCpuMs[source]

gc_wall_ms: cardano_node_tests.utils.model_ekg.GcWallMs[source]

init_cpu_ms: cardano_node_tests.utils.model_ekg.InitCpuMs[source]

init_wall_ms: cardano_node_tests.utils.model_ekg.InitWallMs[source]

max_bytes_slop: cardano_node_tests.utils.model_ekg.MaxBytesSlop[source]

max_bytes_used: cardano_node_tests.utils.model_ekg.MaxBytesUsed[source]

mutator_cpu_ms: cardano_node_tests.utils.model_ekg.MutatorCpuMs[source]

mutator_wall_ms: cardano_node_tests.utils.model_ekg.MutatorWallMs[source]

num_bytes_usage_samples: cardano_node_tests.utils.model_ekg.NumBytesUsageSamples[source]

num_gcs: cardano_node_tests.utils.model_ekg.NumGcs[source]

par_avg_bytes_copied: cardano_node_tests.utils.model_ekg.ParAvgBytesCopied[source]

par_max_bytes_copied: cardano_node_tests.utils.model_ekg.ParMaxBytesCopied[source]

par_tot_bytes_copied: cardano_node_tests.utils.model_ekg.ParTotBytesCopied[source]

peak_megabytes_allocated: cardano_node_tests.utils.model_ekg.PeakMegabytesAllocated[source]

wall_ms: cardano_node_tests.utils.model_ekg.WallMs[source]

class cardano_node_tests.utils.model_ekg.GcCpuMs(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.GcLiveBytes(*, int: cardano_node_tests.utils.model_ekg.Int23)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int23[source]

class cardano_node_tests.utils.model_ekg.GcMajorNum(*, int: cardano_node_tests.utils.model_ekg.Int22)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int22[source]

class cardano_node_tests.utils.model_ekg.GcMinorNum(*, int: cardano_node_tests.utils.model_ekg.Int24)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int24[source]

class cardano_node_tests.utils.model_ekg.GcWallMs(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Gcticks(*, int: cardano_node_tests.utils.model_ekg.Int26)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int26[source]

class cardano_node_tests.utils.model_ekg.Header(*, counter: cardano_node_tests.utils.model_ekg.Counter)[source]

Bases: pydantic.main.BaseModel

counter: cardano_node_tests.utils.model_ekg.Counter[source]

class cardano_node_tests.utils.model_ekg.InitCpuMs(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.InitWallMs(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int1(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int10(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int11(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int12(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int13(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int14(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int15(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int16(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int17(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int18(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int19(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int2(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int20(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int21(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int22(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int23(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int24(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int25(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int26(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int27(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int28(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int29(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int3(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int30(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int4(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int5(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int6(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int7(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int8(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Int9(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.IohkMonitoringVersion(*, val: str, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: str[source]

class cardano_node_tests.utils.model_ekg.MaxBytesSlop(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.MaxBytesUsed(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Mem(*, resident: cardano_node_tests.utils.model_ekg.Resident)[source]

Bases: pydantic.main.BaseModel

resident: cardano_node_tests.utils.model_ekg.Resident[source]

class cardano_node_tests.utils.model_ekg.MempoolBytes(*, int: cardano_node_tests.utils.model_ekg.Int21)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int21[source]

Bases: pydantic.main.BaseModel

Forge: cardano_node_tests.utils.model_ekg.Forge[source]

Mem: cardano_node_tests.utils.model_ekg.Mem[source]

RTS: cardano_node_tests.utils.model_ekg.RTS[source]

Stat: cardano_node_tests.utils.model_ekg.Stat[source]

blockNum: cardano_node_tests.utils.model_ekg.BlockNum[source]

blocksForgedNum: cardano_node_tests.utils.model_ekg.BlocksForgedNum[source]

currentKESPeriod: cardano_node_tests.utils.model_ekg.CurrentKESPeriod[source]

delegMapSize: cardano_node_tests.utils.model_ekg.DelegMapSize[source]

density: cardano_node_tests.utils.model_ekg.Density[source]

epoch: cardano_node_tests.utils.model_ekg.Epoch[source]

mempoolBytes: cardano_node_tests.utils.model_ekg.MempoolBytes[source]

myBlocksUncoupled: cardano_node_tests.utils.model_ekg.MyBlocksUncoupled[source]

nodeIsLeaderNum: cardano_node_tests.utils.model_ekg.NodeIsLeaderNum[source]

nodeStartTime: cardano_node_tests.utils.model_ekg.NodeStartTime[source]

operationalCertificateExpiryKESPeriod: cardano_node_tests.utils.model_ekg.OperationalCertificateExpiryKESPeriod[source]

operationalCertificateStartKESPeriod: cardano_node_tests.utils.model_ekg.OperationalCertificateStartKESPeriod[source]

remainingKESPeriods: cardano_node_tests.utils.model_ekg.RemainingKESPeriods[source]

served: cardano_node_tests.utils.model_ekg.Served[source]

slotInEpoch: cardano_node_tests.utils.model_ekg.SlotInEpoch[source]

slotNum: cardano_node_tests.utils.model_ekg.SlotNum[source]

txsInMempool: cardano_node_tests.utils.model_ekg.TxsInMempool[source]

txsProcessedNum: cardano_node_tests.utils.model_ekg.TxsProcessedNum[source]

utxoSize: cardano_node_tests.utils.model_ekg.UtxoSize[source]

class cardano_node_tests.utils.model_ekg.Model(*, cardano: cardano_node_tests.utils.model_ekg.Cardano, ekg: cardano_node_tests.utils.model_ekg.Ekg, rts: cardano_node_tests.utils.model_ekg.Rts, **extra_data: Any)[source]

Bases: pydantic.main.BaseModel

cardano: cardano_node_tests.utils.model_ekg.Cardano[source]

ekg: cardano_node_tests.utils.model_ekg.Ekg[source]

iohk_monitoring_version: cardano_node_tests.utils.model_ekg.IohkMonitoringVersion[source]

rts: cardano_node_tests.utils.model_ekg.Rts[source]

class cardano_node_tests.utils.model_ekg.MutatorCpuMs(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.MutatorWallMs(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.Mutticks(*, int: cardano_node_tests.utils.model_ekg.Int25)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int25[source]

class cardano_node_tests.utils.model_ekg.MyBlocksUncoupled(*, int: cardano_node_tests.utils.model_ekg.Int10)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int10[source]

class cardano_node_tests.utils.model_ekg.Node(*, metrics: cardano_node_tests.utils.model_ekg.Metrics)[source]

Bases: pydantic.main.BaseModel

metrics: cardano_node_tests.utils.model_ekg.Metrics[source]

class cardano_node_tests.utils.model_ekg.NodeIsLeader(*, int: cardano_node_tests.utils.model_ekg.Int4)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int4[source]

class cardano_node_tests.utils.model_ekg.NodeIsLeaderNum(*, int: cardano_node_tests.utils.model_ekg.Int)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int[source]

class cardano_node_tests.utils.model_ekg.NodeNotLeader(*, int: cardano_node_tests.utils.model_ekg.Int5)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int5[source]

class cardano_node_tests.utils.model_ekg.NodeStartTime(*, int: cardano_node_tests.utils.model_ekg.Int12)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int12[source]

class cardano_node_tests.utils.model_ekg.NumBytesUsageSamples(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.NumGcs(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.OperationalCertificateExpiryKESPeriod(*, int: cardano_node_tests.utils.model_ekg.Int15)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int15[source]

class cardano_node_tests.utils.model_ekg.OperationalCertificateStartKESPeriod(*, int: cardano_node_tests.utils.model_ekg.Int14)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int14[source]

class cardano_node_tests.utils.model_ekg.ParAvgBytesCopied(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.ParMaxBytesCopied(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.ParTotBytesCopied(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.PeakMegabytesAllocated(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.RTS(*, gcMajorNum: cardano_node_tests.utils.model_ekg.GcMajorNum, gcLiveBytes: cardano_node_tests.utils.model_ekg.GcLiveBytes, gcMinorNum: cardano_node_tests.utils.model_ekg.GcMinorNum, mutticks: cardano_node_tests.utils.model_ekg.Mutticks, gcticks: cardano_node_tests.utils.model_ekg.Gcticks)[source]

Bases: pydantic.main.BaseModel

gcLiveBytes: cardano_node_tests.utils.model_ekg.GcLiveBytes[source]

gcMajorNum: cardano_node_tests.utils.model_ekg.GcMajorNum[source]

gcMinorNum: cardano_node_tests.utils.model_ekg.GcMinorNum[source]

gcticks: cardano_node_tests.utils.model_ekg.Gcticks[source]

mutticks: cardano_node_tests.utils.model_ekg.Mutticks[source]

class cardano_node_tests.utils.model_ekg.Real(*, val: str, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: str[source]

class cardano_node_tests.utils.model_ekg.RemainingKESPeriods(*, int: cardano_node_tests.utils.model_ekg.Int16)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int16[source]

class cardano_node_tests.utils.model_ekg.Resident(*, int: cardano_node_tests.utils.model_ekg.Int13)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int13[source]

class cardano_node_tests.utils.model_ekg.Rts(*, gc: cardano_node_tests.utils.model_ekg.Gc)[source]

Bases: pydantic.main.BaseModel

gc: cardano_node_tests.utils.model_ekg.Gc[source]

class cardano_node_tests.utils.model_ekg.Served(*, header: cardano_node_tests.utils.model_ekg.Header)[source]

Bases: pydantic.main.BaseModel

header: cardano_node_tests.utils.model_ekg.Header[source]

class cardano_node_tests.utils.model_ekg.ServerTimestampMs(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

class cardano_node_tests.utils.model_ekg.SlotInEpoch(*, int: cardano_node_tests.utils.model_ekg.Int27)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int27[source]

class cardano_node_tests.utils.model_ekg.SlotNum(*, int: cardano_node_tests.utils.model_ekg.Int28)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int28[source]

class cardano_node_tests.utils.model_ekg.Stat(*, cputicks: cardano_node_tests.utils.model_ekg.Cputicks, threads: cardano_node_tests.utils.model_ekg.Threads)[source]

Bases: pydantic.main.BaseModel

cputicks: cardano_node_tests.utils.model_ekg.Cputicks[source]

threads: cardano_node_tests.utils.model_ekg.Threads[source]

class cardano_node_tests.utils.model_ekg.Threads(*, int: cardano_node_tests.utils.model_ekg.Int30)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int30[source]

class cardano_node_tests.utils.model_ekg.TxsInMempool(*, int: cardano_node_tests.utils.model_ekg.Int17)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int17[source]

class cardano_node_tests.utils.model_ekg.TxsProcessedNum(*, int: cardano_node_tests.utils.model_ekg.Int11)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int11[source]

class cardano_node_tests.utils.model_ekg.UtxoSize(*, int: cardano_node_tests.utils.model_ekg.Int19)[source]

Bases: pydantic.main.BaseModel

int: cardano_node_tests.utils.model_ekg.Int19[source]

class cardano_node_tests.utils.model_ekg.WallMs(*, val: int, type: str)[source]

Bases: pydantic.main.BaseModel

type: str[source]

val: int[source]

cardano_node_tests.utils package

Submodules

cardano_node_tests.utils.artifacts module

cardano_node_tests.utils.cluster_nodes module

cardano_node_tests.utils.cluster_scripts module

cardano_node_tests.utils.clusterlib_utils module

cardano_node_tests.utils.configuration module

cardano_node_tests.utils.dbsync_conn module

cardano_node_tests.utils.dbsync_queries module

cardano_node_tests.utils.dbsync_utils module

cardano_node_tests.utils.helpers module

cardano_node_tests.utils.locking module

cardano_node_tests.utils.logfiles module

cardano_node_tests.utils.model_ekg module

cardano_node_tests.utils.pytest_utils module