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

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]

Save CLI coverage info.

cardano_node_tests.utils.artifacts.save_cluster_artifacts(save_dir: pathlib.Path, state_dir: pathlib.Path) → None

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]

Save info about CLI commands executed by cluster start script.

cardano_node_tests.utils.blockers module

Functionality for checking if an issue is blocked and thus blocking a test.

class cardano_node_tests.utils.blockers.GH(issue: int, repo: str = 'IntersectMBO/cardano-node', fixed_in: str = '', message: str = '', check_on_devel: bool = True)

Bases: object

Methods for working with GitHub issues.

issue

A GitHub issue number.

repo

A repository where the issue belongs to. Default: IntersectMBO/cardano-node.

fixed_in

A version of the project where the issue is fixed. Ignored on unknown projects.

message

A message to be added to blocking outcome.

check_on_devel

A boolean flag indicating if the issue should be checked on devel versions of the project. Default: True.

copy() → cardano_node_tests.utils.blockers.GH

Return a copy of the object.

finish_test() → None

Fail or Xfail test with GitHub issue reference.

is_blocked() → bool

Check if issue is blocked.

cardano_node_tests.utils.blockers.finish_test(issues: Iterable[cardano_node_tests.utils.blockers.GH]) → None

Fail or Xfail test with references to multiple GitHub issues.

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: pathlib.Path, state_dir: pathlib.Path, work_dir: pathlib.Path, instance_num: int, cluster_era: str, tx_era: str, command_era: str)

Bases: object

cluster_era: str

command_era: str

instance_num: int

socket_path: pathlib.Path

state_dir: pathlib.Path

tx_era: str

work_dir: pathlib.Path

class cardano_node_tests.utils.cluster_nodes.ClusterType

Bases: object

Generic cluster type.

LOCAL: Final[str] = 'local'

NODES: ClassVar[Set[str]] = {}

TESTNET: Final[str] = 'testnet'

create_addrs_data(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, destination_dir: Union[str, pathlib.Path] = '.') → Dict[str, Dict[str, Any]]

Create addresses and their keys for usage in tests.

get_cluster_obj(protocol: str = '', tx_era: str = '', slots_offset: int = 0, command_era: str = '') → cardano_clusterlib.clusterlib_klass.ClusterLib

Return instance of ClusterLib (cluster_obj).

test_addr_records: ClassVar[Tuple[str, ...]] = ('user1',)

property testnet_type: str

property uses_shortcut: bool

Check if cluster uses shortcut to go from Byron to last supported era.

class cardano_node_tests.utils.cluster_nodes.LocalCluster

Bases: cardano_node_tests.utils.cluster_nodes.ClusterType

Local cluster type (full cardano mode).

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

create_addrs_data(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, destination_dir: Union[str, pathlib.Path] = '.') → Dict[str, Dict[str, Any]]

Create addresses and their keys for usage in tests.

get_cluster_obj(protocol: str = '', tx_era: str = '', slots_offset: int = 0, command_era: str = '') → cardano_clusterlib.clusterlib_klass.ClusterLib

Return instance of ClusterLib (cluster_obj).

property uses_shortcut: bool

Check if cluster uses shortcut to go from Byron to last supported era.

class cardano_node_tests.utils.cluster_nodes.ServiceStatus(name: str, status: str, pid: Optional[int], uptime: Optional[str], message: str = '')

Bases: object

message: str = ''

name: str

pid: Optional[int]

status: str

uptime: Optional[str]

class cardano_node_tests.utils.cluster_nodes.TestnetCluster

Bases: cardano_node_tests.utils.cluster_nodes.ClusterType

Testnet cluster type (full cardano mode).

NODES: ClassVar[Set[str]] = {'relay1'}

TESTNETS: ClassVar[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'}, 1654041600: {'byron_epochs': 4, 'type': 'preprod'}, 1666656000: {'byron_epochs': 0, 'type': 'preview'}, 1689206400: {'byron_epochs': 0, 'type': 'shelley_qa'}}

create_addrs_data(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, destination_dir: Union[str, pathlib.Path] = '.') → Dict[str, Dict[str, Any]]

Create addresses and their keys for usage in tests.

get_cluster_obj(protocol: str = '', tx_era: str = '', slots_offset: int = 0, command_era: str = '') → cardano_clusterlib.clusterlib_klass.ClusterLib

Return instance of ClusterLib (cluster_obj).

property testnet_type: str

Return testnet type (shelley_qa, etc.).

property uses_shortcut: bool

Check if cluster uses shortcut to go from Byron to last supported era.

class cardano_node_tests.utils.cluster_nodes.Testnets

Bases: object

mainnet: Final[str] = 'mainnet'

preprod: Final[str] = 'preprod'

preview: Final[str] = 'preview'

shelley_qa: Final[str] = 'shelley_qa'

staging: Final[str] = 'staging'

testnet: Final[str] = 'testnet'

cardano_node_tests.utils.cluster_nodes.get_cardano_node_socket_path(instance_num: int, socket_file_name: str = '') → pathlib.Path

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

Get cardano cluster environment.

cardano_node_tests.utils.cluster_nodes.get_cluster_type() → cardano_node_tests.utils.cluster_nodes.ClusterType

Return instance of the cluster type indicated by configuration.

cardano_node_tests.utils.cluster_nodes.get_instance_num() → int

Get cardano cluster instance number.

cardano_node_tests.utils.cluster_nodes.load_addrs_data() → dict

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

Load data for pools existing in the cluster environment.

cardano_node_tests.utils.cluster_nodes.reload_supervisor_config(instance_num: Optional[int] = None) → None

Reload supervisor configuration.

cardano_node_tests.utils.cluster_nodes.restart_all_nodes(instance_num: Optional[int] = None) → None

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

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

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]

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

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

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

Start cluster.

cardano_node_tests.utils.cluster_nodes.start_nodes(node_names: List[str], instance_num: Optional[int] = None) → None

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

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: pathlib.Path, stop_script: pathlib.Path, start_script_args: List[str], dir: pathlib.Path)

Bases: object

dir: pathlib.Path

start_script: pathlib.Path

start_script_args: List[str]

stop_script: pathlib.Path

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

Bases: object

base: int

bft1: int

ekg_bft1: int

ekg_pool1: int

ekg_pool2: int

ekg_pool3: int

ekg_relay1: int

metrics_submit_api: int

node_ports: Tuple[cardano_node_tests.utils.cluster_scripts.NodePorts, ...]

pool1: int

pool2: int

pool3: int

prometheus_bft1: int

prometheus_pool1: int

prometheus_pool2: int

prometheus_pool3: int

prometheus_relay1: int

relay1: int

submit_api: int

supervisor: int

webserver: int

class cardano_node_tests.utils.cluster_scripts.LocalScripts(num_pools: int = - 1)

Bases: cardano_node_tests.utils.cluster_scripts.ScriptsTypes

Scripts for starting local cluster.

copy_scripts_files(destdir: Union[str, pathlib.Path]) → cardano_node_tests.utils.cluster_scripts.StartupFiles

Make copy of cluster scripts files located in this repository.

gen_split_topology_files(destdir: Union[str, pathlib.Path], instance_num: int, offset: int = 0) → None

Generate topology files for split network.

get_instance_ports(instance_num: int) → cardano_node_tests.utils.cluster_scripts.InstancePorts

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

Prepare scripts files for starting and stopping cluster instance.

class cardano_node_tests.utils.cluster_scripts.NodePorts(num: int, node: int, ekg: int, prometheus: int)

Bases: object

ekg: int

node: int

num: int

prometheus: int

class cardano_node_tests.utils.cluster_scripts.ScriptsTypes

Bases: object

Generic cluster scripts.

LOCAL: Final[str] = 'local'

TESTNET: Final[str] = 'testnet'

copy_scripts_files(destdir: Union[str, pathlib.Path]) → cardano_node_tests.utils.cluster_scripts.StartupFiles

Make copy of cluster scripts files.

gen_split_topology_files(destdir: Union[str, pathlib.Path], instance_num: int, offset: int = 0) → None

Generate topology files for split network.

get_instance_ports(instance_num: int) → cardano_node_tests.utils.cluster_scripts.InstancePorts

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

Prepare scripts files for starting and stopping cluster instance.

class cardano_node_tests.utils.cluster_scripts.StartupFiles(start_script: pathlib.Path, genesis_spec: pathlib.Path, config_glob: str)

Bases: object

config_glob: str

genesis_spec: pathlib.Path

start_script: pathlib.Path

class cardano_node_tests.utils.cluster_scripts.TestnetScripts

Bases: cardano_node_tests.utils.cluster_scripts.ScriptsTypes

Scripts for starting a node on testnet.

BOOTSTRAP_CONF: Final[str] = 'testnet_conf'

TESTNET_GLOBS: ClassVar[Tuple[str, ...]] = ('config*.json', 'genesis-*.json', 'topology-*.json', 'dbsync-config.*', 'submit-api-config.*')

copy_scripts_files(destdir: Union[str, pathlib.Path]) → cardano_node_tests.utils.cluster_scripts.StartupFiles

Make copy of cluster scripts files located in this repository.

get_instance_ports(instance_num: int) → cardano_node_tests.utils.cluster_scripts.InstancePorts

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

Prepare scripts files for starting and stopping cluster instance.

There is just one cluster instance running for a given testnet. We keep the instance_num support anyway, as this makes it possible to run multiple testnets on the same machine.

cardano_node_tests.utils.clusterlib_utils module

Utilities that extends the functionality of cardano-clusterlib.

class cardano_node_tests.utils.clusterlib_utils.TokenRecord(token: str, amount: int, issuers_addrs: List[cardano_clusterlib.structs.AddressRecord], token_mint_addr: cardano_clusterlib.structs.AddressRecord, script: pathlib.Path)

Bases: object

amount: int

issuers_addrs: List[cardano_clusterlib.structs.AddressRecord]

script: pathlib.Path

token: str

token_mint_addr: cardano_clusterlib.structs.AddressRecord

class cardano_node_tests.utils.clusterlib_utils.TxMetadata(metadata: dict, aux_data: list)

Bases: object

aux_data: list

metadata: dict

class cardano_node_tests.utils.clusterlib_utils.UpdateProposal(arg: str, value: Any, name: str = '', check_func: Optional[Callable] = None)

Bases: object

arg: str

check_func: Optional[Callable] = None

name: str = ''

value: Any

cardano_node_tests.utils.clusterlib_utils.build_and_submit_tx(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, name_template: str, src_address: str, submit_method: str = '', use_build_cmd: bool = False, txins: Union[List[cardano_clusterlib.structs.UTXOData], Tuple[()]] = (), txouts: Union[List[cardano_clusterlib.structs.TxOut], Tuple[()]] = (), readonly_reference_txins: Union[List[cardano_clusterlib.structs.UTXOData], Tuple[()]] = (), script_txins: Union[List[cardano_clusterlib.structs.ScriptTxIn], Tuple[()]] = (), return_collateral_txouts: Union[List[cardano_clusterlib.structs.TxOut], Tuple[()]] = (), total_collateral_amount: Optional[int] = None, mint: Union[List[cardano_clusterlib.structs.Mint], Tuple[()]] = (), tx_files: Optional[cardano_clusterlib.structs.TxFiles] = None, complex_certs: Union[List[cardano_clusterlib.structs.ComplexCert], Tuple[()]] = (), change_address: str = '', fee_buffer: Optional[int] = None, raw_fee: Optional[int] = None, required_signers: Union[List[Union[str, pathlib.Path]], List[str], List[pathlib.Path], Tuple[()]] = (), required_signer_hashes: Optional[List[str]] = None, withdrawals: Union[List[cardano_clusterlib.structs.TxOut], Tuple[()]] = (), script_withdrawals: Union[List[cardano_clusterlib.structs.ScriptWithdrawal], Tuple[()]] = (), deposit: Optional[int] = None, invalid_hereafter: Optional[int] = None, invalid_before: Optional[int] = None, witness_override: Optional[int] = None, witness_count_add: int = 0, script_valid: bool = True, calc_script_cost_file: Optional[Union[str, pathlib.Path]] = None, join_txouts: bool = True, destination_dir: Union[str, pathlib.Path] = '.', skip_asset_balancing: bool = False) → cardano_clusterlib.structs.TxRawOutput

Build and submit a transaction.

Use use_build_cmd to switch between transaction build and transaction build-raw. Use submit_method to switch between cardano-cli transaction submit and submit-api.

cardano_node_tests.utils.clusterlib_utils.check_pool_data(pool_params: dict, pool_creation_data: cardano_clusterlib.structs.PoolData) → str

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

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

Compare update proposals with actual protocol parameters.

cardano_node_tests.utils.clusterlib_utils.cli_has(command: str) → bool

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]

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, destination_dir: Union[str, pathlib.Path] = '.') → List[cardano_clusterlib.structs.PoolUser]

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]

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

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]

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

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, deposit_amt: int = - 1) → Tuple[cardano_clusterlib.structs.TxRawOutput, cardano_clusterlib.structs.TxRawOutput]

Deregister stake address.

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

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

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]

Get blocksBefore section of ledger state with bech32 encoded pool ids.

cardano_node_tests.utils.clusterlib_utils.get_delegation_state(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib) → dict

Get delegationState section of ledger state.

cardano_node_tests.utils.clusterlib_utils.get_ledger_state(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib) → dict

Return the current ledger state info.

cardano_node_tests.utils.clusterlib_utils.get_plutus_b64(script_file: Union[str, pathlib.Path]) → str

Get base64 encoded binary version of Plutus script from file.

cardano_node_tests.utils.clusterlib_utils.get_pool_state(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, pool_id: str) → cardano_clusterlib.structs.PoolParamsTop

Get pool state using the available command.

cardano_node_tests.utils.clusterlib_utils.get_pparams_update_args(update_proposals: List[cardano_node_tests.utils.clusterlib_utils.UpdateProposal]) → List[str]

Get cli arguments for pparams update action.

cardano_node_tests.utils.clusterlib_utils.get_snapshot_delegations(ledger_snapshot: dict) → Dict[str, List[str]]

Get delegations data from ledger state snapshot.

cardano_node_tests.utils.clusterlib_utils.get_snapshot_rec(ledger_snapshot: dict) → Dict[str, Union[int, list]]

Get uniform record for ledger state snapshot.

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

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

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

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

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, submit_method: str = 'cli', use_build_cmd: bool = False, sign_incrementally: bool = False) → cardano_clusterlib.structs.TxRawOutput

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, submit_method: str = 'cli', use_build_cmd: bool = False, sign_incrementally: bool = False) → cardano_clusterlib.structs.TxRawOutput

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]

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, deposit_amt: int = - 1) → cardano_clusterlib.structs.TxRawOutput

Register stake address.

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

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

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

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

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

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

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

Function for checking a transaction in db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check a transaction in db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_collaterals(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that the Tx collaterals match the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_datum(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that the Tx datum match the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_fee(tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that the Tx fee matches the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_ins(tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that the Tx inputs match the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_mint(tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that the Tx minting matches the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_outs(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that the Tx outputs match the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_reference_inputs(tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that the Tx reference inputs match the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_reference_scripts(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that reference scripts in Tx inputs match the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_required_signers(tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that the Tx required signers match the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_scripts(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that the Tx script hashes match the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_validity(tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that the Tx validity interval match the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_withdrawals(tx_raw_output: cardano_clusterlib.structs.TxRawOutput, response: cardano_node_tests.utils.dbsync_types.TxRecord) → None

Check that the Tx withdrawals match the data from db-sync.

cardano_node_tests.utils.dbsync_check_tx.utxodata2txout(utxodata: Union[cardano_node_tests.utils.dbsync_types.UTxORecord, cardano_clusterlib.structs.UTXOData]) → cardano_clusterlib.structs.TxOut

Convert UTxORecord or UTxOData to clusterlib.TxOut.

cardano_node_tests.utils.dbsync_check_tx.utxorecord2utxodata(utxorecord: cardano_node_tests.utils.dbsync_types.UTxORecord) → cardano_clusterlib.structs.UTXOData

Convert UTxORecord to clusterlib.UTXOData.

cardano_node_tests.utils.dbsync_conn module

Functionality for interacting with db-sync database in postgres.

class cardano_node_tests.utils.dbsync_conn.DBSyncCache

Bases: object

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

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

cardano_node_tests.utils.dbsync_conn.close_all() → None

cardano_node_tests.utils.dbsync_conn.conn() → psycopg2.extensions.connection

cardano_node_tests.utils.dbsync_conn.reconn() → psycopg2.extensions.connection

cardano_node_tests.utils.dbsync_queries module

SQL queries to db-sync database.

class cardano_node_tests.utils.dbsync_queries.ADAPotsDBRow(id: int, slot_no: int, epoch_no: int, treasury: decimal.Decimal, reserves: decimal.Decimal, rewards: decimal.Decimal, utxo: decimal.Decimal, deposits: decimal.Decimal, fees: decimal.Decimal, block_id: int)

Bases: object

block_id: int

deposits: decimal.Decimal

epoch_no: int

fees: decimal.Decimal

id: int

reserves: decimal.Decimal

rewards: decimal.Decimal

slot_no: int

treasury: decimal.Decimal

utxo: decimal.Decimal

class cardano_node_tests.utils.dbsync_queries.ADAStashDBRow(id: int, addr_view: str, cert_index: int, amount: decimal.Decimal, tx_id: int)

Bases: object

addr_view: str

amount: decimal.Decimal

cert_index: int

id: int

tx_id: int

class cardano_node_tests.utils.dbsync_queries.BlockDBRow(id: int, epoch_no: Optional[int], slot_no: Optional[int], epoch_slot_no: Optional[int], block_no: Optional[int], previous_id: Optional[int], tx_count: Optional[int], proto_major: Optional[int], proto_minor: Optional[int], pool_id: Optional[str])

Bases: object

block_no: Optional[int]

epoch_no: Optional[int]

epoch_slot_no: Optional[int]

id: int

pool_id: Optional[str]

previous_id: Optional[int]

proto_major: Optional[int]

proto_minor: Optional[int]

slot_no: Optional[int]

tx_count: Optional[int]

class cardano_node_tests.utils.dbsync_queries.CollateralTxOutDBRow(tx_out_id: int, utxo_ix: int, address: str, value: decimal.Decimal, tx_hash: memoryview)

Bases: object

address: str

tx_hash: memoryview

tx_out_id: int

utxo_ix: int

value: decimal.Decimal

class cardano_node_tests.utils.dbsync_queries.CommitteeDeregistrationDBRow(id: int, tx_id: int, cert_index: int, voting_anchor_id: int, cold_key: memoryview)

Bases: object

cert_index: int

cold_key: memoryview

id: int

tx_id: int

voting_anchor_id: int

class cardano_node_tests.utils.dbsync_queries.CommitteeRegistrationDBRow(id: int, tx_id: int, cert_index: int, cold_key: memoryview, hot_key: memoryview)

Bases: object

cert_index: int

cold_key: memoryview

hot_key: memoryview

id: int

tx_id: int

class cardano_node_tests.utils.dbsync_queries.DatumDBRow(id: int, datum_hash: memoryview, tx_id: int, value: dict, bytes: memoryview)

Bases: object

bytes: memoryview

datum_hash: memoryview

id: int

tx_id: int

value: dict

class cardano_node_tests.utils.dbsync_queries.DrepRegistrationDBRow(id: int, tx_id: int, cert_index: int, deposit: int, drep_hash_id: int, voting_anchor_id: int, hash_raw: memoryview, hash_view: str, has_script: bool)

Bases: object

cert_index: int

deposit: int

drep_hash_id: int

has_script: bool

hash_raw: memoryview

hash_view: str

id: int

tx_id: int

voting_anchor_id: int

class cardano_node_tests.utils.dbsync_queries.EpochDBRow(id: int, out_sum: int, fees: int, tx_count: int, blk_count: int, epoch_number: int)

Bases: object

blk_count: int

epoch_number: int

fees: int

id: int

out_sum: int

tx_count: int

class cardano_node_tests.utils.dbsync_queries.EpochStakeDBRow(id: int, hash: memoryview, view: str, amount: int, epoch_number: int)

Bases: object

amount: int

epoch_number: int

hash: memoryview

id: int

view: str

class cardano_node_tests.utils.dbsync_queries.GovActionProposalDBRow(id: int, tx_id: int, prev_gov_action_proposal: int, deposit: int, return_address: int, expiration: int, voting_anchor_id: int, type: str, description: str, param_proposal: int, ratified_epoch: int, enacted_epoch: int, dropped_epoch: int, expired_epoch: int)

Bases: object

deposit: int

description: str

dropped_epoch: int

enacted_epoch: int

expiration: int

expired_epoch: int

id: int

param_proposal: int

prev_gov_action_proposal: int

ratified_epoch: int

return_address: int

tx_id: int

type: str

voting_anchor_id: int

class cardano_node_tests.utils.dbsync_queries.MetadataDBRow(id: int, key: decimal.Decimal, json: Any, bytes: memoryview, tx_id: int)

Bases: object

bytes: memoryview

id: int

json: Any

key: decimal.Decimal

tx_id: int

class cardano_node_tests.utils.dbsync_queries.ParamProposalDBRow(id: int, epoch_no: int, key: memoryview, min_fee_a: int, min_fee_b: int, max_block_size: int, max_tx_size: int, max_bh_size: int, key_deposit: int, pool_deposit: int, max_epoch: int, optimal_pool_count: int, influence: float, monetary_expand_rate: float, treasury_growth_rate: float, decentralisation: float, entropy: memoryview, protocol_major: int, protocol_minor: int, min_utxo_value: int, min_pool_cost: int, coins_per_utxo_word: int, cost_model_id: int, price_mem: float, price_step: float, max_tx_ex_mem: int, max_tx_ex_steps: int, max_block_ex_mem: int, max_block_ex_steps: int, max_val_size: int, collateral_percent: int, max_collateral_inputs: int, registered_tx_id: int, pvt_motion_no_confidence: int, pvt_committee_normal: int, pvt_committee_no_confidence: int, pvt_hard_fork_initiation: int, dvt_motion_no_confidence: int, dvt_committee_normal: int, dvt_committee_no_confidence: int, dvt_update_to_constitution: int, dvt_hard_fork_initiation: int, dvt_p_p_network_group: int, dvt_p_p_economic_group: int, dvt_p_p_technical_group: int, dvt_p_p_gov_group: int, dvt_treasury_withdrawal: int, committee_min_size: int, committee_max_term_length: int, gov_action_lifetime: int, gov_action_deposit: int, drep_deposit: int, drep_activity: str)

Bases: object

coins_per_utxo_word: int

collateral_percent: int

committee_max_term_length: int

committee_min_size: int

cost_model_id: int

decentralisation: float

drep_activity: str

drep_deposit: int

dvt_committee_no_confidence: int

dvt_committee_normal: int

dvt_hard_fork_initiation: int

dvt_motion_no_confidence: int

dvt_p_p_economic_group: int

dvt_p_p_gov_group: int

dvt_p_p_network_group: int

dvt_p_p_technical_group: int

dvt_treasury_withdrawal: int

dvt_update_to_constitution: int

entropy: memoryview

epoch_no: int

gov_action_deposit: int

gov_action_lifetime: int

id: int

influence: float

key: memoryview

key_deposit: int

max_bh_size: int

max_block_ex_mem: int

max_block_ex_steps: int

max_block_size: int

max_collateral_inputs: int

max_epoch: int

max_tx_ex_mem: int

max_tx_ex_steps: int

max_tx_size: int

max_val_size: int

min_fee_a: int

min_fee_b: int

min_pool_cost: int

min_utxo_value: int

monetary_expand_rate: float

optimal_pool_count: int

pool_deposit: int

price_mem: float

price_step: float

protocol_major: int

protocol_minor: int

pvt_committee_no_confidence: int

pvt_committee_normal: int

pvt_hard_fork_initiation: int

pvt_motion_no_confidence: int

registered_tx_id: int

treasury_growth_rate: float

class cardano_node_tests.utils.dbsync_queries.PoolDataDBRow(id: int, hash: memoryview, view: str, cert_index: int, vrf_key_hash: memoryview, pledge: int, reward_addr: memoryview, reward_addr_view: str, active_epoch_no: int, meta_id: int, margin: decimal.Decimal, fixed_cost: int, registered_tx_id: int, metadata_url: str, metadata_hash: memoryview, owner_stake_address_id: int, owner: memoryview, ipv4: str, ipv6: str, dns_name: str, port: int, retire_cert_index: int, retire_announced_tx_id: int, retiring_epoch: int)

Bases: object

active_epoch_no: int

cert_index: int

dns_name: str

fixed_cost: int

hash: memoryview

id: int

ipv4: str

ipv6: str

margin: decimal.Decimal

meta_id: int

metadata_hash: memoryview

metadata_url: str

owner: memoryview

owner_stake_address_id: int

pledge: int

port: int

registered_tx_id: int

retire_announced_tx_id: int

retire_cert_index: int

retiring_epoch: int

reward_addr: memoryview

reward_addr_view: str

view: str

vrf_key_hash: memoryview

class cardano_node_tests.utils.dbsync_queries.PoolOffChainDataDBRow(id: int, ticker_name: str, hash: memoryview, json: dict, bytes: memoryview, pmr_id: int)

Bases: object

bytes: memoryview

hash: memoryview

id: int

json: dict

pmr_id: int

ticker_name: str

class cardano_node_tests.utils.dbsync_queries.PoolOffChainFetchErrorDBRow(id: int, pmr_id: int, fetch_error: str, retry_count: int)

Bases: object

fetch_error: str

id: int

pmr_id: int

retry_count: int

class cardano_node_tests.utils.dbsync_queries.PotTransferDBRow(id: int, cert_index: int, treasury: decimal.Decimal, reserves: decimal.Decimal, tx_id: int)

Bases: object

cert_index: int

id: int

reserves: decimal.Decimal

treasury: decimal.Decimal

tx_id: int

class cardano_node_tests.utils.dbsync_queries.RedeemerDBRow(id: int, tx_id: int, unit_mem: int, unit_steps: int, fee: int, purpose: str, script_hash: memoryview, value: dict)

Bases: object

fee: int

id: int

purpose: str

script_hash: memoryview

tx_id: int

unit_mem: int

unit_steps: int

value: dict

class cardano_node_tests.utils.dbsync_queries.RewardDBRow(address: str, type: str, amount: decimal.Decimal, earned_epoch: int, spendable_epoch: int, pool_id: Optional[str] = '')

Bases: object

address: str

amount: decimal.Decimal

earned_epoch: int

pool_id: Optional[str] = ''

spendable_epoch: int

type: str

class cardano_node_tests.utils.dbsync_queries.SchemaVersion

Bases: object

Query and cache db-sync schema version.

classmethod stages() → cardano_node_tests.utils.dbsync_queries.SchemaVersionStages

class cardano_node_tests.utils.dbsync_queries.SchemaVersionStages(one: int, two: int, three: int)

Bases: object

one: int

three: int

two: int

class cardano_node_tests.utils.dbsync_queries.ScriptDBRow(id: int, tx_id: int, hash: memoryview, type: str, serialised_size: Optional[int])

Bases: object

hash: memoryview

id: int

serialised_size: Optional[int]

tx_id: int

type: str

class cardano_node_tests.utils.dbsync_queries.StakeAddrDBRow(id: int, view: str, tx_id: int)

Bases: object

id: int

tx_id: int

view: str

class cardano_node_tests.utils.dbsync_queries.StakeDelegDBRow(tx_id: int, active_epoch_no: Optional[int], pool_id: Optional[str], address: Optional[str])

Bases: object

active_epoch_no: Optional[int]

address: Optional[str]

pool_id: Optional[str]

tx_id: int

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

Bases: object

block_id: int

block_index: int

collateral_count: int

collateral_out_count: int

deposit: int

extra_key_witness_count: int

fee: decimal.Decimal

invalid_before: Optional[decimal.Decimal]

invalid_hereafter: Optional[decimal.Decimal]

ma_tx_mint_id: Optional[int]

ma_tx_mint_name: Optional[memoryview]

ma_tx_mint_policy: Optional[memoryview]

ma_tx_mint_quantity: Optional[decimal.Decimal]

ma_tx_out_id: Optional[int]

ma_tx_out_name: Optional[memoryview]

ma_tx_out_policy: Optional[memoryview]

ma_tx_out_quantity: Optional[decimal.Decimal]

metadata_count: int

out_sum: decimal.Decimal

pot_transfer_count: int

redeemer_count: int

reference_input_count: int

reserve_count: int

script_count: int

size: int

stake_deleg_count: int

stake_dereg_count: int

stake_reg_count: int

treasury_count: int

tx_hash: memoryview

tx_id: int

tx_out_addr: str

tx_out_addr_has_script: bool

tx_out_data_hash: Optional[memoryview]

tx_out_id: int

tx_out_inline_datum_hash: Optional[memoryview]

tx_out_reference_script_hash: Optional[memoryview]

tx_out_tx_id: int

tx_out_value: decimal.Decimal

utxo_ix: int

withdrawal_count: int

class cardano_node_tests.utils.dbsync_queries.TxInDBRow(tx_out_id: int, utxo_ix: int, address: str, value: decimal.Decimal, tx_hash: memoryview, reference_script_hash: Optional[memoryview], reference_script_json: Optional[dict], reference_script_bytes: Optional[memoryview], reference_script_type: Optional[str], ma_tx_out_id: Optional[int], ma_tx_out_policy: Optional[memoryview], ma_tx_out_name: Optional[memoryview], ma_tx_out_quantity: Optional[decimal.Decimal])

Bases: object

address: str

ma_tx_out_id: Optional[int]

ma_tx_out_name: Optional[memoryview]

ma_tx_out_policy: Optional[memoryview]

ma_tx_out_quantity: Optional[decimal.Decimal]

reference_script_bytes: Optional[memoryview]

reference_script_hash: Optional[memoryview]

reference_script_json: Optional[dict]

reference_script_type: Optional[str]

tx_hash: memoryview

tx_out_id: int

utxo_ix: int

value: decimal.Decimal

class cardano_node_tests.utils.dbsync_queries.TxInNoMADBRow(tx_out_id: int, utxo_ix: int, address: str, value: decimal.Decimal, tx_hash: memoryview, reference_script_hash: Optional[memoryview], reference_script_json: Optional[dict], reference_script_bytes: Optional[memoryview], reference_script_type: Optional[str])

Bases: object

address: str

reference_script_bytes: Optional[memoryview]

reference_script_hash: Optional[memoryview]

reference_script_json: Optional[dict]

reference_script_type: Optional[str]

tx_hash: memoryview

tx_out_id: int

utxo_ix: int

value: decimal.Decimal

class cardano_node_tests.utils.dbsync_queries.UTxODBRow(tx_hash: memoryview, utxo_ix: int, payment_address: str, stake_address: str, has_script: bool, value: int, data_hash: Optional[memoryview])

Bases: object

data_hash: Optional[memoryview]

has_script: bool

payment_address: str

stake_address: str

tx_hash: memoryview

utxo_ix: int

value: int

class cardano_node_tests.utils.dbsync_queries.WithdrawalDBRow(tx_id: int, address: str, amount: int)

Bases: object

address: str

amount: int

tx_id: int

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

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]

Query ADA pots record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_address_instant_reward(address: str, epoch_from: int = 0, epoch_to: int = 99999999) → Generator[cardano_node_tests.utils.dbsync_queries.RewardDBRow, None, None]

Query instant reward records for stake address 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]

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]

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]

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]

Query transaction collateral txouts in db-sync.

cardano_node_tests.utils.dbsync_queries.query_committee_deregistration(cold_key: str) → Generator[cardano_node_tests.utils.dbsync_queries.CommitteeDeregistrationDBRow, None, None]

Query committee registration in db-sync.

cardano_node_tests.utils.dbsync_queries.query_committee_registration(cold_key: str) → Generator[cardano_node_tests.utils.dbsync_queries.CommitteeRegistrationDBRow, None, None]

Query committee registration in db-sync.

cardano_node_tests.utils.dbsync_queries.query_cost_model() → Dict[str, Dict[str, Any]]

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]

Query datum record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_drep_registration(drep_hash: str, drep_deposit: int = 2000000) → Generator[cardano_node_tests.utils.dbsync_queries.DrepRegistrationDBRow, None, None]

Query drep registration 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]

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]

Query epoch stake record for a pool in db-sync.

cardano_node_tests.utils.dbsync_queries.query_extra_key_witness(txhash: str) → Generator[memoryview, None, None]

Query extra key witness records in db-sync.

cardano_node_tests.utils.dbsync_queries.query_gov_action_proposal(txhash: str = '', type: str = '') → Generator[cardano_node_tests.utils.dbsync_queries.GovActionProposalDBRow, None, None]

Query gov_action_proposal table in db-sync.

If type is provided txhash will be ignored.

cardano_node_tests.utils.dbsync_queries.query_off_chain_pool_data(pool_id_bech32: str) → Generator[cardano_node_tests.utils.dbsync_queries.PoolOffChainDataDBRow, None, None]

Query Off_Chain_Pool_Data record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_off_chain_pool_fetch_error(pool_id_bech32: str) → Generator[cardano_node_tests.utils.dbsync_queries.PoolOffChainFetchErrorDBRow, None, None]

Query Off_Chain_Pool_Fetch_Error record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_param_proposal(txhash: str = '') → cardano_node_tests.utils.dbsync_queries.ParamProposalDBRow

Query param proposal record in db-sync.

If txhash is not provided the query will return the last record available.

cardano_node_tests.utils.dbsync_queries.query_pool_data(pool_id_bech32: str) → Generator[cardano_node_tests.utils.dbsync_queries.PoolDataDBRow, None, None]

Query pool data record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_redeemers(txhash: str) → Generator[cardano_node_tests.utils.dbsync_queries.RedeemerDBRow, None, None]

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]

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]

Query transaction scripts in db-sync.

cardano_node_tests.utils.dbsync_queries.query_table_names() → List[str]

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]

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]

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]

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]

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]

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]

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]

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]

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]

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]

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]

Query UTxOs for payment address in db-sync.

cardano_node_tests.utils.dbsync_types module

Types used in db-sync related functions.

class cardano_node_tests.utils.dbsync_types.ADAStashRecord(address: str, cert_index: int, amount: int)

Bases: object

address: str

amount: int

cert_index: int

class cardano_node_tests.utils.dbsync_types.CommitteeDeregistrationRecord(id: int, tx_id: int, cert_index: int, voting_anchor_id: int, cold_key: str)

Bases: object

cert_index: int

cold_key: str

id: int

tx_id: int

voting_anchor_id: int

class cardano_node_tests.utils.dbsync_types.CommitteeRegistrationRecord(id: int, tx_id: int, cert_index: int, cold_key: str, hot_key: str)

Bases: object

cert_index: int

cold_key: str

hot_key: str

id: int

tx_id: int

class cardano_node_tests.utils.dbsync_types.DelegationRecord(address: str, pool_id: str, active_epoch_no: int)

Bases: object

active_epoch_no: int

address: str

pool_id: str

class cardano_node_tests.utils.dbsync_types.DrepRegistrationRecord(id: int, tx_id: int, cert_index: int, deposit: int, drep_hash_id: int, voting_anchor_id: int, hash_hex: str, hash_bech32: str, has_script: bool)

Bases: object

cert_index: int

deposit: int

drep_hash_id: int

has_script: bool

hash_bech32: str

hash_hex: str

id: int

tx_id: int

voting_anchor_id: int

class cardano_node_tests.utils.dbsync_types.GetUTxORecord(utxo_hash: str, utxo_ix: int, has_script: bool, amount: int, data_hash: str)

Bases: object

amount: int

data_hash: str

has_script: bool

utxo_hash: str

utxo_ix: int

class cardano_node_tests.utils.dbsync_types.MetadataRecord(key: int, json: Any, bytes: memoryview)

Bases: object

bytes: memoryview

json: Any

key: int

class cardano_node_tests.utils.dbsync_types.PaymentAddrRecord(payment_address: str, stake_address: Optional[str], amount_sum: int, utxos: List[cardano_node_tests.utils.dbsync_types.GetUTxORecord])

Bases: object

amount_sum: int

payment_address: str

stake_address: Optional[str]

utxos: List[cardano_node_tests.utils.dbsync_types.GetUTxORecord]

class cardano_node_tests.utils.dbsync_types.PoolDataRecord(id: int, hash: str, view: str, cert_index: int, vrf_key_hash: str, pledge: int, reward_addr: str, active_epoch_no: int, meta_id: int, margin: float, fixed_cost: int, registered_tx_id: int, metadata_url: str, metadata_hash: str, owners: List[str], relays: List[Dict[str, Dict[str, Any]]], retire_cert_index: int, retire_announced_tx_id: int, retiring_epoch: int)

Bases: object

active_epoch_no: int

cert_index: int

fixed_cost: int

hash: str

id: int

margin: float

meta_id: int

metadata_hash: str

metadata_url: str

owners: List[str]

pledge: int

registered_tx_id: int

relays: List[Dict[str, Dict[str, Any]]]

retire_announced_tx_id: int

retire_cert_index: int

retiring_epoch: int

reward_addr: str

view: str

vrf_key_hash: str

class cardano_node_tests.utils.dbsync_types.PotTransferRecord(treasury: int, reserves: int)

Bases: object

reserves: int

treasury: int

class cardano_node_tests.utils.dbsync_types.RedeemerRecord(unit_mem: int, unit_steps: int, fee: int, purpose: str, script_hash: str, value: dict)

Bases: object

fee: int

purpose: str

script_hash: str

unit_mem: int

unit_steps: int

value: dict

class cardano_node_tests.utils.dbsync_types.RewardEpochRecord(amount: int, earned_epoch: int, spendable_epoch: int, type: str, pool_id: str)

Bases: object

amount: int

earned_epoch: int

pool_id: str

spendable_epoch: int

type: str

class cardano_node_tests.utils.dbsync_types.RewardRecord(address: str, rewards: List[cardano_node_tests.utils.dbsync_types.RewardEpochRecord], reward_sum: int)

Bases: object

address: str

reward_sum: int

rewards: List[cardano_node_tests.utils.dbsync_types.RewardEpochRecord]

class cardano_node_tests.utils.dbsync_types.ScriptRecord(hash: str, type: str, serialised_size: int)

Bases: object

hash: str

serialised_size: int

type: str

class cardano_node_tests.utils.dbsync_types.TxPrelimRecord(utxo_out: List[cardano_node_tests.utils.dbsync_types.UTxORecord], ma_utxo_out: List[cardano_node_tests.utils.dbsync_types.UTxORecord], mint_utxo_out: List[cardano_node_tests.utils.dbsync_types.UTxORecord], last_row: cardano_node_tests.utils.dbsync_queries.TxDBRow)

Bases: object

last_row: cardano_node_tests.utils.dbsync_queries.TxDBRow

ma_utxo_out: List[cardano_node_tests.utils.dbsync_types.UTxORecord]

mint_utxo_out: List[cardano_node_tests.utils.dbsync_types.UTxORecord]

utxo_out: List[cardano_node_tests.utils.dbsync_types.UTxORecord]

class cardano_node_tests.utils.dbsync_types.TxRecord(tx_id: int, tx_hash: str, block_id: int, block_index: int, out_sum: int, fee: int, deposit: int, size: int, invalid_before: Optional[int], invalid_hereafter: Optional[int], txins: List[cardano_node_tests.utils.dbsync_types.UTxORecord], txouts: List[cardano_node_tests.utils.dbsync_types.UTxORecord], mint: List[cardano_node_tests.utils.dbsync_types.UTxORecord], collaterals: List[cardano_node_tests.utils.dbsync_types.UTxORecord], collateral_outputs: List[cardano_clusterlib.structs.UTXOData], reference_inputs: List[cardano_node_tests.utils.dbsync_types.UTxORecord], scripts: List[cardano_node_tests.utils.dbsync_types.ScriptRecord], redeemers: List[cardano_node_tests.utils.dbsync_types.RedeemerRecord], metadata: List[cardano_node_tests.utils.dbsync_types.MetadataRecord], reserve: List[cardano_node_tests.utils.dbsync_types.ADAStashRecord], treasury: List[cardano_node_tests.utils.dbsync_types.ADAStashRecord], pot_transfers: List[cardano_node_tests.utils.dbsync_types.PotTransferRecord], stake_registration: List[str], stake_deregistration: List[str], stake_delegation: List[cardano_node_tests.utils.dbsync_types.DelegationRecord], withdrawals: List[cardano_clusterlib.structs.TxOut], extra_key_witness: List[str])

Bases: object

block_id: int

block_index: int

collateral_outputs: List[cardano_clusterlib.structs.UTXOData]

collaterals: List[cardano_node_tests.utils.dbsync_types.UTxORecord]

deposit: int

extra_key_witness: List[str]

fee: int

invalid_before: Optional[int]

invalid_hereafter: Optional[int]

metadata: List[cardano_node_tests.utils.dbsync_types.MetadataRecord]

mint: List[cardano_node_tests.utils.dbsync_types.UTxORecord]

out_sum: int

pot_transfers: List[cardano_node_tests.utils.dbsync_types.PotTransferRecord]

redeemers: List[cardano_node_tests.utils.dbsync_types.RedeemerRecord]

reference_inputs: List[cardano_node_tests.utils.dbsync_types.UTxORecord]

reserve: List[cardano_node_tests.utils.dbsync_types.ADAStashRecord]

scripts: List[cardano_node_tests.utils.dbsync_types.ScriptRecord]

size: int

stake_delegation: List[cardano_node_tests.utils.dbsync_types.DelegationRecord]

stake_deregistration: List[str]

stake_registration: List[str]

treasury: List[cardano_node_tests.utils.dbsync_types.ADAStashRecord]

tx_hash: str

tx_id: int

txins: List[cardano_node_tests.utils.dbsync_types.UTxORecord]

txouts: List[cardano_node_tests.utils.dbsync_types.UTxORecord]

withdrawals: List[cardano_clusterlib.structs.TxOut]

class cardano_node_tests.utils.dbsync_types.UTxORecord(utxo_hash: str, utxo_ix: int, amount: int, address: str, coin: str = 'lovelace', decoded_coin: str = '', datum_hash: str = '', inline_datum_hash: str = '', inline_datum: Union[str, dict, NoneType] = None, reference_script: Optional[dict] = None, reference_script_hash: str = '')

Bases: object

address: str

amount: int

coin: str = 'lovelace'

datum_hash: str = ''

decoded_coin: str = ''

inline_datum: Optional[Union[str, dict]] = None

inline_datum_hash: str = ''

reference_script: Optional[dict] = None

reference_script_hash: str = ''

utxo_hash: str

utxo_ix: int

cardano_node_tests.utils.dbsync_utils module

Functionality for interacting with db-sync.

cardano_node_tests.utils.dbsync_utils.check_address_reward(address: str, epoch_from: int = 0, epoch_to: int = 99999999) → cardano_node_tests.utils.dbsync_types.RewardRecord

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_committee_member_deregistration(cc_member_cold_key: str) → Optional[cardano_node_tests.utils.dbsync_types.CommitteeDeregistrationRecord]

Check committee member deregistration in db-sync.

cardano_node_tests.utils.dbsync_utils.check_committee_member_registration(cc_member_cold_key: str, committee_state: Dict[str, Any]) → Optional[cardano_node_tests.utils.dbsync_types.CommitteeRegistrationRecord]

Check committee member registration in db-sync.

cardano_node_tests.utils.dbsync_utils.check_drep_deregistration(drep: cardano_node_tests.utils.governance_utils.DRepRegistration) → Optional[cardano_node_tests.utils.dbsync_types.DrepRegistrationRecord]

Check drep deregistration in db-sync.

cardano_node_tests.utils.dbsync_utils.check_drep_registration(drep: cardano_node_tests.utils.governance_utils.DRepRegistration, drep_state: List[List[Dict[str, Any]]]) → Optional[cardano_node_tests.utils.dbsync_types.DrepRegistrationRecord]

Check drep registration in db-sync.

cardano_node_tests.utils.dbsync_utils.check_param_proposal(protocol_params: dict) → Optional[cardano_node_tests.utils.dbsync_queries.ParamProposalDBRow]

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_types.RedeemerRecord, cost_record: Dict[str, Any]) → None

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_types.RedeemerRecord], cost_records: List[Dict[str, Any]]) → None

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_types.PoolDataRecord]

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_types.PoolDataRecord]

Check pool retirement in db-sync.

cardano_node_tests.utils.dbsync_utils.check_pool_off_chain_data(ledger_pool_data: dict, pool_id: str) → cardano_node_tests.utils.dbsync_queries.PoolOffChainDataDBRow

Check comparison for pool off chain data between ledger and db-sync.

cardano_node_tests.utils.dbsync_utils.check_pool_off_chain_fetch_error(ledger_pool_data: dict, pool_id: str) → cardano_node_tests.utils.dbsync_queries.PoolOffChainFetchErrorDBRow

Check expected error on PoolOffChainFetchError.

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_types.TxRecord]

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_types.TxRecord]

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_types.RewardRecord

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_committee_member(cold_key: str) → Optional[cardano_node_tests.utils.dbsync_types.CommitteeRegistrationRecord]

Get committee member data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_deregistered_committee_member(cold_key: str) → Optional[cardano_node_tests.utils.dbsync_types.CommitteeDeregistrationRecord]

Get deregistered committee member data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_drep(drep_hash: str, drep_deposit: int) → Optional[cardano_node_tests.utils.dbsync_types.DrepRegistrationRecord]

Get drep data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_pool_data(pool_id_bech32: str) → Optional[cardano_node_tests.utils.dbsync_types.PoolDataRecord]

Get pool data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_prelim_tx_record(txhash: str) → cardano_node_tests.utils.dbsync_types.TxPrelimRecord

Get first batch of transaction data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_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_types.TxRecord]

Get a transaction data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_tx_record(txhash: str) → cardano_node_tests.utils.dbsync_types.TxRecord

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_types.TxRecord

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_node_tests.utils.dbsync_types.UTxORecord]

Get txins of a transaction from db-sync.

cardano_node_tests.utils.dbsync_utils.get_utxo(address: str) → cardano_node_tests.utils.dbsync_types.PaymentAddrRecord

Return UTxO info for payment address from db-sync.

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

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.faucet module

cardano_node_tests.utils.faucet.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]

Send amount from faucet addr to all dst_addrs.

cardano_node_tests.utils.faucet.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

Send amount from all src_addrs to faucet_addr.

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

cardano_node_tests.utils.gh_issue module

Functionality for working with GitHub issues.

class cardano_node_tests.utils.gh_issue.GHIssue(number: int, repo: str)

Bases: object

GitHub issue.

TOKEN: ClassVar[Optional[str]] = 'token_XXXXXXXXXXXXXXXXXXXX'

get_state() → Optional[str]

Get issue state.

property github: Optional[github.MainClass.Github]

is_closed() → bool

Check if issue is closed.

issue_cache: ClassVar[Dict[str, str]] = {}

property url: str

cardano_node_tests.utils.governance_setup module

class cardano_node_tests.utils.governance_setup.DefaultGovernance(dreps_reg: List[cardano_node_tests.utils.governance_utils.DRepRegistration], drep_delegators: List[cardano_clusterlib.structs.PoolUser], cc_members: List[cardano_clusterlib.structs.CCMember], pools_cold: List[cardano_clusterlib.structs.ColdKeyPair])

Bases: object

cc_members: List[cardano_clusterlib.structs.CCMember]

drep_delegators: List[cardano_clusterlib.structs.PoolUser]

dreps_reg: List[cardano_node_tests.utils.governance_utils.DRepRegistration]

pools_cold: List[cardano_clusterlib.structs.ColdKeyPair]

cardano_node_tests.utils.governance_setup.auth_cc_members(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, cc_members: List[cardano_clusterlib.structs.CCMember], name_template: str, payment_addr: cardano_clusterlib.structs.AddressRecord) → None

Authorize the original CC members.

cardano_node_tests.utils.governance_setup.create_dreps(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, payment_addr: cardano_clusterlib.structs.AddressRecord, pool_users: List[cardano_clusterlib.structs.PoolUser], destination_dir: Union[str, pathlib.Path] = '.') → Tuple[List[cardano_node_tests.utils.governance_utils.DRepRegistration], List[cardano_clusterlib.structs.PoolUser]]

cardano_node_tests.utils.governance_setup.create_vote_stake(cluster_manager: cardano_node_tests.cluster_management.manager.ClusterManager, cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, destination_dir: Union[str, pathlib.Path] = '.') → List[cardano_clusterlib.structs.PoolUser]

cardano_node_tests.utils.governance_setup.get_default_governance(cluster_manager: cardano_node_tests.cluster_management.manager.ClusterManager, cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib) → cardano_node_tests.utils.governance_setup.DefaultGovernance

cardano_node_tests.utils.governance_setup.load_committee(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib) → List[cardano_clusterlib.structs.CCMember]

cardano_node_tests.utils.governance_setup.refresh_cc_keys(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, cc_members: List[cardano_clusterlib.structs.CCMember], governance_data: cardano_node_tests.utils.governance_setup.DefaultGovernance) → cardano_node_tests.utils.governance_setup.DefaultGovernance

Refresh ho certs for original CC members.

cardano_node_tests.utils.governance_setup.reinstate_committee(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, governance_data: cardano_node_tests.utils.governance_setup.DefaultGovernance, name_template: str, pool_user: cardano_clusterlib.structs.PoolUser) → None

Reinstate the original CC members.

cardano_node_tests.utils.governance_setup.save_default_governance(dreps_reg: List[cardano_node_tests.utils.governance_utils.DRepRegistration], drep_delegators: List[cardano_clusterlib.structs.PoolUser], cc_members: List[cardano_clusterlib.structs.CCMember], pools_cold: List[cardano_clusterlib.structs.ColdKeyPair]) → cardano_node_tests.utils.governance_setup.DefaultGovernance

Save governance data to a pickle, so it can be reused.

This needs to be called either under a file lock (FileLockIfXdist), or in a test that locked whole governance for itself.

cardano_node_tests.utils.governance_setup.setup(cluster_manager: cardano_node_tests.cluster_management.manager.ClusterManager, cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, destination_dir: Union[str, pathlib.Path] = '.') → cardano_node_tests.utils.governance_setup.DefaultGovernance

cardano_node_tests.utils.governance_utils module

Utilities for Conway governance.

class cardano_node_tests.utils.governance_utils.ActionTags(value)

Bases: enum.Enum

An enumeration.

INFO_ACTION = 'InfoAction'

NEW_CONSTITUTION = 'NewConstitution'

NO_CONFIDENCE = 'NoConfidence'

PARAMETER_CHANGE = 'ParameterChange'

TREASURY_WITHDRAWALS = 'TreasuryWithdrawals'

UPDATE_COMMITTEE = 'UpdateCommittee'

class cardano_node_tests.utils.governance_utils.CCMemberAuth(auth_cert: pathlib.Path, cold_key_pair: cardano_clusterlib.structs.KeyPair, hot_key_pair: cardano_clusterlib.structs.KeyPair, key_hash: str)

Bases: object

auth_cert: pathlib.Path

cold_key_pair: cardano_clusterlib.structs.KeyPair

hot_key_pair: cardano_clusterlib.structs.KeyPair

key_hash: str

class cardano_node_tests.utils.governance_utils.DRepRegistration(registration_cert: pathlib.Path, key_pair: cardano_clusterlib.structs.KeyPair, drep_id: str, deposit: int)

Bases: object

deposit: int

drep_id: str

key_pair: cardano_clusterlib.structs.KeyPair

registration_cert: pathlib.Path

class cardano_node_tests.utils.governance_utils.PrevActionRec(txid: str, ix: int)

Bases: object

ix: int

txid: str

class cardano_node_tests.utils.governance_utils.PrevGovActionIds(value)

Bases: enum.Enum

An enumeration.

COMMITTEE = 'Committee'

CONSTITUTION = 'Constitution'

HARDFORK = 'HardFork'

PPARAM_UPDATE = 'PParamUpdate'

class cardano_node_tests.utils.governance_utils.StakeDelegation(spo: int, drep: int, total_lovelace: int)

Bases: object

drep: int

spo: int

total_lovelace: int

cardano_node_tests.utils.governance_utils.check_action_view(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, action_data: Union[cardano_clusterlib.structs.ActionConstitution, cardano_clusterlib.structs.ActionInfo, cardano_clusterlib.structs.ActionNoConfidence, cardano_clusterlib.structs.ActionPParamsUpdate, cardano_clusterlib.structs.ActionUpdateCommittee, cardano_clusterlib.structs.ActionTreasuryWithdrawal], return_addr_vkey_hash: str = '', recv_addr_vkey_hash: str = '') → None

Check governance action view output.

cardano_node_tests.utils.governance_utils.check_drep_delegation(deleg_state: dict, drep_id: str, stake_addr_hash: str) → None

cardano_node_tests.utils.governance_utils.check_drep_stake_distribution(distrib_state: List[list], drep_id: str, min_amount: int) → None

cardano_node_tests.utils.governance_utils.check_vote_view(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, vote_data: Union[cardano_clusterlib.structs.VoteCC, cardano_clusterlib.structs.VoteDrep, cardano_clusterlib.structs.VoteSPO]) → None

Check governance vote view output.

cardano_node_tests.utils.governance_utils.get_cc_member_auth_record(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, name_template: str, destination_dir: Union[str, pathlib.Path] = '.') → cardano_node_tests.utils.governance_utils.CCMemberAuth

Get Constitutional Committee Members key authorization record.

cardano_node_tests.utils.governance_utils.get_delegated_stake(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib) → cardano_node_tests.utils.governance_utils.StakeDelegation

Get total stake delegated to SPOs and DReps.

cardano_node_tests.utils.governance_utils.get_drep_cred_name(drep_id: str) → str

cardano_node_tests.utils.governance_utils.get_drep_reg_record(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, name_template: str, deposit_amt: int = - 1, drep_metadata_url: str = '', drep_metadata_hash: str = '', destination_dir: Union[str, pathlib.Path] = '.') → cardano_node_tests.utils.governance_utils.DRepRegistration

Get DRep registration record.

cardano_node_tests.utils.governance_utils.get_prev_action(action_type: cardano_node_tests.utils.governance_utils.PrevGovActionIds, gov_state: Dict[str, Any]) → cardano_node_tests.utils.governance_utils.PrevActionRec

cardano_node_tests.utils.governance_utils.get_vote_str(vote: cardano_clusterlib.consts.Votes) → str

cardano_node_tests.utils.governance_utils.lookup_expired_actions(gov_state: Dict[str, Any], action_txid: str, action_ix: int = 0) → Dict[str, Any]

cardano_node_tests.utils.governance_utils.lookup_proposal(gov_state: Dict[str, Any], action_txid: str, action_ix: int = 0) → Dict[str, Any]

cardano_node_tests.utils.governance_utils.lookup_ratified_actions(gov_state: Dict[str, Any], action_txid: str, action_ix: int = 0) → Dict[str, Any]

cardano_node_tests.utils.governance_utils.wait_delayed_ratification(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib) → None

Wait until ratification is no longer delayed.

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

Call a function and cache its return value.

Warning

The function arguments are not considered when caching the result. Therefore, this decorator should be used only for functions without arguments or for functions with constant arguments.

cardano_node_tests.utils.helpers.change_cwd(dir_path: Union[str, pathlib.Path]) → Iterator[Union[str, pathlib.Path]]

Change and restore CWD - context manager.

cardano_node_tests.utils.helpers.check_dir_arg(dir_path: str) → Optional[pathlib.Path]

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]

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

Return file checksum.

cardano_node_tests.utils.helpers.decode_bech32(bech32: str) → str

Convert from bech32 string.

cardano_node_tests.utils.helpers.encode_bech32(prefix: str, data: str) → str

Convert to bech32 string.

cardano_node_tests.utils.helpers.environ(env: dict) → Iterator[None]

Temporarily set environment variables and restore previous environment afterwards.

cardano_node_tests.utils.helpers.flatten(iterable: typing.Iterable, ltypes: type[typing.Iterable[typing.Any]] = <class 'collections.abc.Iterable'>) → Generator

Flatten an irregular (arbitrarily nested) iterable of iterables.

cardano_node_tests.utils.helpers.get_current_commit() → str

cardano_node_tests.utils.helpers.get_current_line_str() → str

Get filename#lineno of current line.

NOTE: this will not work correctly if called from context manager.

cardano_node_tests.utils.helpers.get_eof_offset(infile: pathlib.Path) → int

Return position of the current end of the file.

cardano_node_tests.utils.helpers.get_line_str_from_frame(frame: frame) → str

cardano_node_tests.utils.helpers.get_rand_str(length: int = 8) → str

Return random string.

cardano_node_tests.utils.helpers.get_timestamped_rand_str(rand_str_length: int = 4) → str

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

Return link to the current line in GitHub.

cardano_node_tests.utils.helpers.ignore_interrupt() → Iterator[None]

Ignore the KeyboardInterrupt signal.

cardano_node_tests.utils.helpers.is_in_interval(num1: float, num2: float, frac: float = 0.1) → bool

Check that the num1 is in the interval defined by num2 and its fraction.

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

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("--foo", [1, 2, 3])
['--foo', '1', '--foo', '2', '--foo', '3']

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

Run command.

cardano_node_tests.utils.helpers.run_in_bash(command: str, workdir: Union[str, pathlib.Path] = '') → bytes

Run command(s) in bash.

cardano_node_tests.utils.helpers.tool_has(command: str) → bool

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.write_json(out_file: Union[str, pathlib.Path], content: dict) → Union[str, pathlib.Path]

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: pathlib.Path, seek: int, timestamp: float)

Bases: object

logfile: pathlib.Path

seek: int

timestamp: float

cardano_node_tests.utils.logfiles.add_ignore_rule(files_glob: str, regex: str, ignore_file_id: str, skip_after: float = 0.0) → None

Add ignore rule for expected errors.

Parameters

• files_glob – A glob matching files that the regex should apply to.

• regex – A regex that should be ignored.

• ignore_file_id –

  The id of a ignore file the ignore rule will be added to.

  NOTE: When ignore_file_id matches pytest-xdist worker id (the worker_id fixture), the rule will be deleted during the test teardown.

• skip_after –

  The time in seconds after which the rule will expire. This is to avoid reporting the ignored errors in subsequent tests. It can take several seconds for the errors to appear in log files and we don’t want to wait for them after each test.

  NOTE: The rule will expire only when there are no yet to be searched log messages that were created before the skip_after time.

cardano_node_tests.utils.logfiles.clean_ignore_rules(ignore_file_id: str) → None

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]], worker_id: str) → Iterator[None]

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.

• worker_id – The id of the pytest-xdist worker (the worker_id fixture) that the test is running on.

cardano_node_tests.utils.logfiles.framework_logger() → logging.Logger

Get logger for the framework.log file.

The logger is configured per worker. It can be used for logging (and later reporting) events like a failure to start a cluster instance.

cardano_node_tests.utils.logfiles.get_framework_log_path() → pathlib.Path

cardano_node_tests.utils.logfiles.get_logfiles_errors() → str

Get errors found in cluster artifacts.

cardano_node_tests.utils.logfiles.search_cluster_logs() → List[Tuple[pathlib.Path, str]]

Search cluster logs for errors.

cardano_node_tests.utils.logfiles.search_framework_log() → List[Tuple[pathlib.Path, str]]

Search framework log for errors.

cardano_node_tests.utils.logfiles.search_supervisord_logs() → List[Tuple[pathlib.Path, str]]

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)

Bases: pydantic.main.BaseModel

int: Int7

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int7, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.BlockNum(*, int: cardano_node_tests.utils.model_ekg.Int1)

Bases: pydantic.main.BaseModel

int: Int1

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int1, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.BlocksForgedNum(*, int: cardano_node_tests.utils.model_ekg.Int2)

Bases: pydantic.main.BaseModel

int: Int2

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int2, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.BytesAllocated(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.BytesCopied(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Cardano(*, node: cardano_node_tests.utils.model_ekg.Node)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'node': FieldInfo(annotation=Node, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

node: Node

class cardano_node_tests.utils.model_ekg.Counter(*, int: cardano_node_tests.utils.model_ekg.Int20)

Bases: pydantic.main.BaseModel

int: Int20

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int20, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.CpuMs(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Cputicks(*, int: cardano_node_tests.utils.model_ekg.Int29)

Bases: pydantic.main.BaseModel

int: Int29

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int29, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.CumulativeBytesUsed(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.CurrentBytesSlop(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.CurrentBytesUsed(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.CurrentKESPeriod(*, int: cardano_node_tests.utils.model_ekg.Int9)

Bases: pydantic.main.BaseModel

int: Int9

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int9, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.DelegMapSize(*, int: cardano_node_tests.utils.model_ekg.Int18)

Bases: pydantic.main.BaseModel

int: Int18

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int18, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.Density(*, real: cardano_node_tests.utils.model_ekg.Real)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'real': FieldInfo(annotation=Real, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

real: Real

class cardano_node_tests.utils.model_ekg.Ekg(*, server_timestamp_ms: cardano_node_tests.utils.model_ekg.ServerTimestampMs)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'server_timestamp_ms': FieldInfo(annotation=ServerTimestampMs, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

server_timestamp_ms: ServerTimestampMs

class cardano_node_tests.utils.model_ekg.Epoch(*, int: cardano_node_tests.utils.model_ekg.Int3)

Bases: pydantic.main.BaseModel

int: Int3

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int3, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

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)

Bases: pydantic.main.BaseModel

adopted: Adopted

forge_about_to_lead: ForgeAboutToLead

forged: Forged

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'adopted': FieldInfo(annotation=Adopted, required=True), 'forge_about_to_lead': FieldInfo(annotation=ForgeAboutToLead, required=True, alias='forge-about-to-lead', alias_priority=2), 'forged': FieldInfo(annotation=Forged, required=True), 'node_is_leader': FieldInfo(annotation=NodeIsLeader, required=True, alias='node-is-leader', alias_priority=2), 'node_not_leader': FieldInfo(annotation=NodeNotLeader, required=True, alias='node-not-leader', alias_priority=2)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

node_is_leader: NodeIsLeader

node_not_leader: NodeNotLeader

class cardano_node_tests.utils.model_ekg.ForgeAboutToLead(*, int: cardano_node_tests.utils.model_ekg.Int8)

Bases: pydantic.main.BaseModel

int: Int8

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int8, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.Forged(*, int: cardano_node_tests.utils.model_ekg.Int6)

Bases: pydantic.main.BaseModel

int: Int6

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int6, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.Gc(*, bytes_allocated: cardano_node_tests.utils.model_ekg.BytesAllocated, mutator_cpu_ms: cardano_node_tests.utils.model_ekg.MutatorCpuMs, gc_wall_ms: cardano_node_tests.utils.model_ekg.GcWallMs, peak_megabytes_allocated: cardano_node_tests.utils.model_ekg.PeakMegabytesAllocated, par_tot_bytes_copied: cardano_node_tests.utils.model_ekg.ParTotBytesCopied, cumulative_bytes_used: cardano_node_tests.utils.model_ekg.CumulativeBytesUsed, par_max_bytes_copied: cardano_node_tests.utils.model_ekg.ParMaxBytesCopied, init_wall_ms: cardano_node_tests.utils.model_ekg.InitWallMs, max_bytes_used: cardano_node_tests.utils.model_ekg.MaxBytesUsed, bytes_copied: cardano_node_tests.utils.model_ekg.BytesCopied, wall_ms: cardano_node_tests.utils.model_ekg.WallMs, cpu_ms: cardano_node_tests.utils.model_ekg.CpuMs, current_bytes_used: cardano_node_tests.utils.model_ekg.CurrentBytesUsed, gc_cpu_ms: cardano_node_tests.utils.model_ekg.GcCpuMs, mutator_wall_ms: cardano_node_tests.utils.model_ekg.MutatorWallMs, par_avg_bytes_copied: cardano_node_tests.utils.model_ekg.ParAvgBytesCopied, num_gcs: cardano_node_tests.utils.model_ekg.NumGcs, current_bytes_slop: cardano_node_tests.utils.model_ekg.CurrentBytesSlop, num_bytes_usage_samples: cardano_node_tests.utils.model_ekg.NumBytesUsageSamples, max_bytes_slop: cardano_node_tests.utils.model_ekg.MaxBytesSlop, init_cpu_ms: cardano_node_tests.utils.model_ekg.InitCpuMs)

Bases: pydantic.main.BaseModel

bytes_allocated: BytesAllocated

bytes_copied: BytesCopied

cpu_ms: CpuMs

cumulative_bytes_used: CumulativeBytesUsed

current_bytes_slop: CurrentBytesSlop

current_bytes_used: CurrentBytesUsed

gc_cpu_ms: GcCpuMs

gc_wall_ms: GcWallMs

init_cpu_ms: InitCpuMs

init_wall_ms: InitWallMs

max_bytes_slop: MaxBytesSlop

max_bytes_used: MaxBytesUsed

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'bytes_allocated': FieldInfo(annotation=BytesAllocated, required=True), 'bytes_copied': FieldInfo(annotation=BytesCopied, required=True), 'cpu_ms': FieldInfo(annotation=CpuMs, required=True), 'cumulative_bytes_used': FieldInfo(annotation=CumulativeBytesUsed, required=True), 'current_bytes_slop': FieldInfo(annotation=CurrentBytesSlop, required=True), 'current_bytes_used': FieldInfo(annotation=CurrentBytesUsed, required=True), 'gc_cpu_ms': FieldInfo(annotation=GcCpuMs, required=True), 'gc_wall_ms': FieldInfo(annotation=GcWallMs, required=True), 'init_cpu_ms': FieldInfo(annotation=InitCpuMs, required=True), 'init_wall_ms': FieldInfo(annotation=InitWallMs, required=True), 'max_bytes_slop': FieldInfo(annotation=MaxBytesSlop, required=True), 'max_bytes_used': FieldInfo(annotation=MaxBytesUsed, required=True), 'mutator_cpu_ms': FieldInfo(annotation=MutatorCpuMs, required=True), 'mutator_wall_ms': FieldInfo(annotation=MutatorWallMs, required=True), 'num_bytes_usage_samples': FieldInfo(annotation=NumBytesUsageSamples, required=True), 'num_gcs': FieldInfo(annotation=NumGcs, required=True), 'par_avg_bytes_copied': FieldInfo(annotation=ParAvgBytesCopied, required=True), 'par_max_bytes_copied': FieldInfo(annotation=ParMaxBytesCopied, required=True), 'par_tot_bytes_copied': FieldInfo(annotation=ParTotBytesCopied, required=True), 'peak_megabytes_allocated': FieldInfo(annotation=PeakMegabytesAllocated, required=True), 'wall_ms': FieldInfo(annotation=WallMs, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

mutator_cpu_ms: MutatorCpuMs

mutator_wall_ms: MutatorWallMs

num_bytes_usage_samples: NumBytesUsageSamples

num_gcs: NumGcs

par_avg_bytes_copied: ParAvgBytesCopied

par_max_bytes_copied: ParMaxBytesCopied

par_tot_bytes_copied: ParTotBytesCopied

peak_megabytes_allocated: PeakMegabytesAllocated

wall_ms: WallMs

class cardano_node_tests.utils.model_ekg.GcCpuMs(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.GcLiveBytes(*, int: cardano_node_tests.utils.model_ekg.Int23)

Bases: pydantic.main.BaseModel

int: Int23

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int23, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.GcMajorNum(*, int: cardano_node_tests.utils.model_ekg.Int22)

Bases: pydantic.main.BaseModel

int: Int22

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int22, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.GcMinorNum(*, int: cardano_node_tests.utils.model_ekg.Int24)

Bases: pydantic.main.BaseModel

int: Int24

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int24, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.GcWallMs(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Gcticks(*, int: cardano_node_tests.utils.model_ekg.Int26)

Bases: pydantic.main.BaseModel

int: Int26

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int26, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.Header(*, counter: cardano_node_tests.utils.model_ekg.Counter)

Bases: pydantic.main.BaseModel

counter: Counter

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'counter': FieldInfo(annotation=Counter, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.InitCpuMs(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.InitWallMs(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int1(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int10(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int11(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int12(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int13(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int14(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int15(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int16(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int17(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int18(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int19(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int2(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int20(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int21(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int22(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int23(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int24(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int25(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int26(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int27(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int28(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int29(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int3(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int30(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int4(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int5(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int6(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int7(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int8(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Int9(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.IohkMonitoringVersion(*, val: str, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: str

class cardano_node_tests.utils.model_ekg.MaxBytesSlop(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.MaxBytesUsed(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Mem(*, resident: cardano_node_tests.utils.model_ekg.Resident)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'resident': FieldInfo(annotation=Resident, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

resident: Resident

class cardano_node_tests.utils.model_ekg.MempoolBytes(*, int: cardano_node_tests.utils.model_ekg.Int21)

Bases: pydantic.main.BaseModel

int: Int21

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int21, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.Metrics(*, nodeIsLeaderNum: cardano_node_tests.utils.model_ekg.NodeIsLeaderNum, blockNum: cardano_node_tests.utils.model_ekg.BlockNum, blocksForgedNum: cardano_node_tests.utils.model_ekg.BlocksForgedNum, epoch: cardano_node_tests.utils.model_ekg.Epoch, Forge: cardano_node_tests.utils.model_ekg.Forge, currentKESPeriod: cardano_node_tests.utils.model_ekg.CurrentKESPeriod, myBlocksUncoupled: cardano_node_tests.utils.model_ekg.MyBlocksUncoupled, txsProcessedNum: cardano_node_tests.utils.model_ekg.TxsProcessedNum, density: cardano_node_tests.utils.model_ekg.Density, nodeStartTime: cardano_node_tests.utils.model_ekg.NodeStartTime, Mem: cardano_node_tests.utils.model_ekg.Mem, operationalCertificateStartKESPeriod: cardano_node_tests.utils.model_ekg.OperationalCertificateStartKESPeriod, operationalCertificateExpiryKESPeriod: cardano_node_tests.utils.model_ekg.OperationalCertificateExpiryKESPeriod, remainingKESPeriods: cardano_node_tests.utils.model_ekg.RemainingKESPeriods, txsInMempool: cardano_node_tests.utils.model_ekg.TxsInMempool, delegMapSize: cardano_node_tests.utils.model_ekg.DelegMapSize, utxoSize: cardano_node_tests.utils.model_ekg.UtxoSize, served: cardano_node_tests.utils.model_ekg.Served, mempoolBytes: cardano_node_tests.utils.model_ekg.MempoolBytes, RTS: cardano_node_tests.utils.model_ekg.RTS, slotInEpoch: cardano_node_tests.utils.model_ekg.SlotInEpoch, slotNum: cardano_node_tests.utils.model_ekg.SlotNum, Stat: cardano_node_tests.utils.model_ekg.Stat)

Bases: pydantic.main.BaseModel

Forge: Forge

Mem: Mem

RTS: RTS

Stat: Stat

blockNum: BlockNum

blocksForgedNum: BlocksForgedNum

currentKESPeriod: CurrentKESPeriod

delegMapSize: DelegMapSize

density: Density

epoch: Epoch

mempoolBytes: MempoolBytes

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'Forge': FieldInfo(annotation=Forge, required=True), 'Mem': FieldInfo(annotation=Mem, required=True), 'RTS': FieldInfo(annotation=RTS, required=True), 'Stat': FieldInfo(annotation=Stat, required=True), 'blockNum': FieldInfo(annotation=BlockNum, required=True), 'blocksForgedNum': FieldInfo(annotation=BlocksForgedNum, required=True), 'currentKESPeriod': FieldInfo(annotation=CurrentKESPeriod, required=True), 'delegMapSize': FieldInfo(annotation=DelegMapSize, required=True), 'density': FieldInfo(annotation=Density, required=True), 'epoch': FieldInfo(annotation=Epoch, required=True), 'mempoolBytes': FieldInfo(annotation=MempoolBytes, required=True), 'myBlocksUncoupled': FieldInfo(annotation=MyBlocksUncoupled, required=True), 'nodeIsLeaderNum': FieldInfo(annotation=NodeIsLeaderNum, required=True), 'nodeStartTime': FieldInfo(annotation=NodeStartTime, required=True), 'operationalCertificateExpiryKESPeriod': FieldInfo(annotation=OperationalCertificateExpiryKESPeriod, required=True), 'operationalCertificateStartKESPeriod': FieldInfo(annotation=OperationalCertificateStartKESPeriod, required=True), 'remainingKESPeriods': FieldInfo(annotation=RemainingKESPeriods, required=True), 'served': FieldInfo(annotation=Served, required=True), 'slotInEpoch': FieldInfo(annotation=SlotInEpoch, required=True), 'slotNum': FieldInfo(annotation=SlotNum, required=True), 'txsInMempool': FieldInfo(annotation=TxsInMempool, required=True), 'txsProcessedNum': FieldInfo(annotation=TxsProcessedNum, required=True), 'utxoSize': FieldInfo(annotation=UtxoSize, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

myBlocksUncoupled: MyBlocksUncoupled

nodeIsLeaderNum: NodeIsLeaderNum

nodeStartTime: NodeStartTime

operationalCertificateExpiryKESPeriod: OperationalCertificateExpiryKESPeriod

operationalCertificateStartKESPeriod: OperationalCertificateStartKESPeriod

remainingKESPeriods: RemainingKESPeriods

served: Served

slotInEpoch: SlotInEpoch

slotNum: SlotNum

txsInMempool: TxsInMempool

txsProcessedNum: TxsProcessedNum

utxoSize: UtxoSize

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)

Bases: pydantic.main.BaseModel

cardano: Cardano

ekg: Ekg

iohk_monitoring_version: IohkMonitoringVersion

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'cardano': FieldInfo(annotation=Cardano, required=True), 'ekg': FieldInfo(annotation=Ekg, required=True), 'iohk_monitoring_version': FieldInfo(annotation=IohkMonitoringVersion, required=True, alias='iohk-monitoring version', alias_priority=2), 'rts': FieldInfo(annotation=Rts, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

rts: Rts

class cardano_node_tests.utils.model_ekg.MutatorCpuMs(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.MutatorWallMs(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.Mutticks(*, int: cardano_node_tests.utils.model_ekg.Int25)

Bases: pydantic.main.BaseModel

int: Int25

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int25, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.MyBlocksUncoupled(*, int: cardano_node_tests.utils.model_ekg.Int10)

Bases: pydantic.main.BaseModel

int: Int10

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int10, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.Node(*, metrics: cardano_node_tests.utils.model_ekg.Metrics)

Bases: pydantic.main.BaseModel

metrics: Metrics

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'metrics': FieldInfo(annotation=Metrics, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.NodeIsLeader(*, int: cardano_node_tests.utils.model_ekg.Int4)

Bases: pydantic.main.BaseModel

int: Int4

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int4, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.NodeIsLeaderNum(*, int: cardano_node_tests.utils.model_ekg.Int)

Bases: pydantic.main.BaseModel

int: Int

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.NodeNotLeader(*, int: cardano_node_tests.utils.model_ekg.Int5)

Bases: pydantic.main.BaseModel

int: Int5

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int5, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.NodeStartTime(*, int: cardano_node_tests.utils.model_ekg.Int12)

Bases: pydantic.main.BaseModel

int: Int12

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int12, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.NumBytesUsageSamples(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.NumGcs(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.OperationalCertificateExpiryKESPeriod(*, int: cardano_node_tests.utils.model_ekg.Int15)

Bases: pydantic.main.BaseModel

int: Int15

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int15, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.OperationalCertificateStartKESPeriod(*, int: cardano_node_tests.utils.model_ekg.Int14)

Bases: pydantic.main.BaseModel

int: Int14

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int14, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.ParAvgBytesCopied(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.ParMaxBytesCopied(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.ParTotBytesCopied(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.PeakMegabytesAllocated(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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)

Bases: pydantic.main.BaseModel

gcLiveBytes: GcLiveBytes

gcMajorNum: GcMajorNum

gcMinorNum: GcMinorNum

gcticks: Gcticks

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'gcLiveBytes': FieldInfo(annotation=GcLiveBytes, required=True), 'gcMajorNum': FieldInfo(annotation=GcMajorNum, required=True), 'gcMinorNum': FieldInfo(annotation=GcMinorNum, required=True), 'gcticks': FieldInfo(annotation=Gcticks, required=True), 'mutticks': FieldInfo(annotation=Mutticks, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

mutticks: Mutticks

class cardano_node_tests.utils.model_ekg.Real(*, val: str, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: str

class cardano_node_tests.utils.model_ekg.RemainingKESPeriods(*, int: cardano_node_tests.utils.model_ekg.Int16)

Bases: pydantic.main.BaseModel

int: Int16

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int16, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.Resident(*, int: cardano_node_tests.utils.model_ekg.Int13)

Bases: pydantic.main.BaseModel

int: Int13

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int13, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.Rts(*, gc: cardano_node_tests.utils.model_ekg.Gc)

Bases: pydantic.main.BaseModel

gc: Gc

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'gc': FieldInfo(annotation=Gc, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.Served(*, header: cardano_node_tests.utils.model_ekg.Header)

Bases: pydantic.main.BaseModel

header: Header

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'header': FieldInfo(annotation=Header, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.ServerTimestampMs(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

class cardano_node_tests.utils.model_ekg.SlotInEpoch(*, int: cardano_node_tests.utils.model_ekg.Int27)

Bases: pydantic.main.BaseModel

int: Int27

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int27, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.SlotNum(*, int: cardano_node_tests.utils.model_ekg.Int28)

Bases: pydantic.main.BaseModel

int: Int28

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int28, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.Stat(*, cputicks: cardano_node_tests.utils.model_ekg.Cputicks, threads: cardano_node_tests.utils.model_ekg.Threads)

Bases: pydantic.main.BaseModel

cputicks: Cputicks

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'cputicks': FieldInfo(annotation=Cputicks, required=True), 'threads': FieldInfo(annotation=Threads, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

threads: Threads

class cardano_node_tests.utils.model_ekg.Threads(*, int: cardano_node_tests.utils.model_ekg.Int30)

Bases: pydantic.main.BaseModel

int: Int30

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int30, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.TxsInMempool(*, int: cardano_node_tests.utils.model_ekg.Int17)

Bases: pydantic.main.BaseModel

int: Int17

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int17, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.TxsProcessedNum(*, int: cardano_node_tests.utils.model_ekg.Int11)

Bases: pydantic.main.BaseModel

int: Int11

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int11, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.UtxoSize(*, int: cardano_node_tests.utils.model_ekg.Int19)

Bases: pydantic.main.BaseModel

int: Int19

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'int': FieldInfo(annotation=Int19, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class cardano_node_tests.utils.model_ekg.WallMs(*, val: int, type: str)

Bases: pydantic.main.BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'type': FieldInfo(annotation=str, required=True), 'val': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

cardano_node_tests.utils.poll_utils module

Functions for working with SPO polls.

class cardano_node_tests.utils.poll_utils.PollFiles(poll: pathlib.Path, metadata: pathlib.Path)

Bases: object

metadata: pathlib.Path

poll: pathlib.Path

cardano_node_tests.utils.poll_utils.answer_poll(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, poll_file: pathlib.Path, answer: int, name_template: str) → pathlib.Path

Answer a poll and return the answer file.

cardano_node_tests.utils.poll_utils.create_poll(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, question: str, answers: List[str], name_template: str) → cardano_node_tests.utils.poll_utils.PollFiles

Create a poll and return the poll and metadata files.

cardano_node_tests.utils.poll_utils.verify_poll(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, poll_file: pathlib.Path, tx_signed: pathlib.Path) → Tuple[str, ...]

Verify an answer to the poll.

cardano_node_tests.utils.pytest_utils module

class cardano_node_tests.utils.pytest_utils.PytestTest(test_function: str, test_file: pathlib.Path, full: str, test_class: str = '', test_params: str = '', stage: str = '')

Bases: object

full: str

stage: str = ''

test_class: str = ''

test_file: pathlib.Path

test_function: str

test_params: str = ''

cardano_node_tests.utils.pytest_utils.get_current_test() → cardano_node_tests.utils.pytest_utils.PytestTest

Get components (test file, test name, etc.) of current pytest test.

cardano_node_tests.utils.requirements module

Functionality for tracking execution of external requirements.

class cardano_node_tests.utils.requirements.GroupsKnown

Bases: object

CHANG_US: Final[str] = 'chang_us'

class cardano_node_tests.utils.requirements.Req(id: str, group: str = '', url: str = '')

Bases: object

Methods for working with external requirements.

id

An identification of the requirement.

group

A group of the requirement.

url

An url of the requirement.

failure() → bool

start(url: str = '') → bool

success() → bool

class cardano_node_tests.utils.requirements.Statuses

Bases: object

FAILURE: Final[str] = 'failure'

PARTIAL_SUCCESS: Final[str] = 'partial_success'

SUCCESS: Final[str] = 'success'

UNCOVERED: Final[str] = 'uncovered'

cardano_node_tests.utils.requirements.collect_executed_req(base_dir: pathlib.Path) → dict

Collect executed requirements.

cardano_node_tests.utils.requirements.get_mapped_req(mapping: pathlib.Path, executed_req: dict) → dict

Get mapped requirements.

cardano_node_tests.utils.slots_offset module

cardano_node_tests.utils.slots_offset.get_slots_offset(genesis_byron: pathlib.Path, genesis_shelley: pathlib.Path, shelley_start: str = '', byron_epochs: int = 1) → int

Get offset of slots between Byron and Shelley eras.

Byron and Shelley eras can have different epoch and slot lengths. The slots offset is needed in order to be able to calculate what slot number to expect in certain point of time (e.g. epoch boundary).

When no values for shelley_start and byron_epochs are provided, assume it’s local cluster where Shelley era starts after a single Byron epoch.

Parameters

• genesis_byron – Path to Byron genesis file.

• genesis_shelley – Path to Shelley genesis file.

• shelley_start – Time of start of Shelley era.

• byron_epochs – Number of Byron epochs.

cardano_node_tests.utils.submit_api module

Utilities for cardano-submit-api REST service.

exception cardano_node_tests.utils.submit_api.SubmitApiError

Bases: Exception

class cardano_node_tests.utils.submit_api.SubmitApiOut(txid: str, response: requests.models.Response)

Bases: object

response: requests.models.Response

txid: str

cardano_node_tests.utils.submit_api.is_running() → bool

Check if cardano-submit-api REST service is running.

cardano_node_tests.utils.submit_api.post_cbor(cbor_file: Union[str, pathlib.Path], url: str) → requests.models.Response

Post binary CBOR representation of Tx to cardano-submit-api service on url.

cardano_node_tests.utils.submit_api.submit_tx(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_file: Union[str, pathlib.Path], txins: List[cardano_clusterlib.structs.UTXOData], wait_blocks: int = 2) → None

Submit a transaction, resubmit if the transaction didn’t make it to the chain.

Parameters

• cluster_obj – An instance of clusterlib.ClusterLib.

• tx_file – A path to signed transaction file.

• txins – An iterable of clusterlib.UTXOData, specifying input UTxOs.

• wait_blocks – A number of new blocks to wait for (default = 2).

cardano_node_tests.utils.submit_api.submit_tx_bare(tx_file: Union[str, pathlib.Path]) → cardano_node_tests.utils.submit_api.SubmitApiOut

Submit a signed Tx using cardano-submit-api service.

cardano_node_tests.utils.submit_api.tx2cbor(tx_file: Union[str, pathlib.Path], destination_dir: Union[str, pathlib.Path] = '.') → pathlib.Path

Convert signed Tx to binary CBOR.

cardano_node_tests.utils.submit_utils module

class cardano_node_tests.utils.submit_utils.SubmitMethods

Bases: object

API: Final[str] = 'api'

CLI: Final[str] = 'cli'

cardano_node_tests.utils.submit_utils.is_submit_api_available() → bool

Check if cardano-submit-api is available.

cardano_node_tests.utils.submit_utils.submit_tx(submit_method: str, cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_file: Union[str, pathlib.Path], txins: List[cardano_clusterlib.structs.UTXOData], wait_blocks: int = 2) → None

Submit a transaction using the selected method.

Parameters

• submit_method – A method to use for submitting the transaction.

• cluster_obj – An instance of clusterlib.ClusterLib.

• tx_file – A path to signed transaction file.

• txins – An iterable of clusterlib.UTXOData, specifying input UTxOs.

• wait_blocks – A number of new blocks to wait for (default = 2).

cardano_node_tests.utils.temptools module

class cardano_node_tests.utils.temptools.PytestTempDirs

Bases: object

Pytest temporary directories that are used accross the framework.

The class is initialized in conftest.py where we have access to the tmp_path_factory fixture.

classmethod init(tmp_path_factory: _pytest.tmpdir.TempPathFactory) → None

pytest_root_tmp: ClassVar[Optional[pathlib.Path]] = None

pytest_shared_tmp: ClassVar[Optional[pathlib.Path]] = None

pytest_worker_tmp: ClassVar[Optional[pathlib.Path]] = None

cardano_node_tests.utils.temptools.get_basetemp() → pathlib.Path

Return base temporary directory for tests artifacts.

cardano_node_tests.utils.temptools.get_pytest_root_tmp() → pathlib.Path

Return root of the Pytest temporary directory for a single Pytest run.

cardano_node_tests.utils.temptools.get_pytest_shared_tmp() → pathlib.Path

Return shared temporary directory for a single Pytest run.

Temporary directory that can be shared by multiple Pytest workers, e.g. for creating lock files.

cardano_node_tests.utils.temptools.get_pytest_worker_tmp() → pathlib.Path

Return Pytest temporary directory for the current worker.

When running pytest with multiple workers, each worker has it’s own base temporary directory inside the “root” temporary directory.

cardano_node_tests.utils.testnet_cleanup module

Cleanup a testnet with the help of testing artifacts.

• withdraw rewards

• deregister stake addresses

• return funds to faucet

cardano_node_tests.utils.testnet_cleanup.cleanup(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, location: Union[str, pathlib.Path]) → None

Cleanup a testnet with the help of testing artifacts.

cardano_node_tests.utils.testnet_cleanup.create_addr_record(addr_file: pathlib.Path) → cardano_clusterlib.structs.AddressRecord

Return a clusterlib.AddressRecord.

cardano_node_tests.utils.testnet_cleanup.deregister_stake_addr(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, pool_user: cardano_clusterlib.structs.PoolUser, name_template: str) → None

Deregister stake address.

cardano_node_tests.utils.testnet_cleanup.find_files(location: Union[str, pathlib.Path]) → Generator[pathlib.Path, None, None]

Find all ‘*.addr’ files in given location and it’s subdirectories.

cardano_node_tests.utils.testnet_cleanup.group_files(file_paths: Generator[pathlib.Path, None, None]) → List[List[pathlib.Path]]

Group payment address files with corresponding stake address files.

These need to be processed together - funds are transferred from payment address after the stake address was deregistered.

cardano_node_tests.utils.testnet_cleanup.return_funds_to_faucet(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, src_addr: cardano_clusterlib.structs.AddressRecord, faucet_address: str, tx_name: str) → None

Send funds from src_addr to faucet_address.

cardano_node_tests.utils.testnet_cleanup.withdraw_reward(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, stake_addr_record: cardano_clusterlib.structs.AddressRecord, dst_addr_record: cardano_clusterlib.structs.AddressRecord, name_template: str) → None

Withdraw rewards to payment address.

cardano_node_tests.utils.tx_view module

Checks for transaction view CLI command.

cardano_node_tests.utils.tx_view.check_tx_view(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_raw_output: cardano_clusterlib.structs.TxRawOutput) → Dict[str, Any]

Check output of the transaction view command.

cardano_node_tests.utils.tx_view.load_raw(tx_view: str) → dict

Load tx view output as YAML.

cardano_node_tests.utils.tx_view.load_tx_view(cluster_obj: cardano_clusterlib.clusterlib_klass.ClusterLib, tx_body_file: pathlib.Path) → Dict[str, Any]

cardano_node_tests.utils.types module

cardano_node_tests.utils.versions module

Cardano node version, cluster era, transaction era, db-sync version.

class cardano_node_tests.utils.versions.Versions

Bases: object

Cluster era, transaction era, node version info.

ALLEGRA: Final[int] = 3

ALONZO: Final[int] = 6

BABBAGE: Final[int] = 8

BYRON: Final[int] = 1

CONWAY: Final[int] = 9

DEFAULT_CLUSTER_ERA: Final[int] = 8

DEFAULT_TX_ERA: Final[int] = 8

LAST_KNOWN_ERA: Final[int] = 9

LATEST_DBSYNC_RELEASE_VER: Final[packaging.version.Version] = <Version('13.2.0.1')>

LATEST_NODE_RELEASE_REV: Final[str] = '0d98405a60d57e1c8e13406d51cce0e34356bd64'

LATEST_NODE_RELEASE_VER: Final[packaging.version.Version] = <Version('8.9.0')>

MAP: ClassVar[Dict[int, str]] = {1: 'byron', 2: 'shelley', 3: 'allegra', 4: 'mary', 6: 'alonzo', 8: 'babbage', 9: 'conway'}

MARY: Final[int] = 4

SHELLEY: Final[int] = 2

get_cardano_cli_version() → dict

Return version info for cardano-cli.

get_cardano_node_version() → dict

Return version info for cardano-node.

get_dbsync_version() → dict

Return version info for db-sync.

Module contents