cardano_node_tests.utils package

Submodules

cardano_node_tests.utils.artifacts module

Functionality for collecting testing artifacts.

cardano_node_tests.utils.artifacts.save_cli_coverage(*, cluster_obj: ClusterLib, pytest_config: Config) → Path | None

Save CLI coverage info.

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

Save cluster artifacts (logs, certs, etc.).

cardano_node_tests.utils.artifacts.save_start_script_coverage(*, log_file: Path, pytest_config: Config) → Path | None

Save info about CLI commands executed by cluster start script.

Returns:

Path to the saved coverage log file, or None when coverage collection is disabled, the log file doesn’t exist, or the copy failed.

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

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. On projects other than cardano-node, cardano-cli and cardano-db-sync, it is the cardano-node version into which the fix was integrated.

message

A message to be added to blocking outcome.

copy() → GH

Return a copy of the object.

finish_test(force_blocked: bool = False) → None

Fail or Xfail test with GitHub issue reference.

Never returns - the test outcome is always set via pytest.xfail or pytest.fail.

Parameters:

force_blocked – Treat the issue as blocked without checking its state.

Raises:

ValueError – If the GitHub issue doesn’t exist or is not accessible. Cannot happen with force_blocked, as the issue state is not checked.

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

Fail or Xfail test with references to multiple GitHub issues.

Never returns - the test outcome is always set via pytest.xfail or pytest.fail.

Parameters:

issues – GitHub issues to report. Must not be empty.

Raises:

ValueError – If no issues were provided, or if a referenced GitHub issue doesn’t exist or is not accessible.

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, command_era: str)

Bases: object

cluster_era: str

command_era: str

instance_num: int

socket_path: Path

state_dir: Path

work_dir: Path

class cardano_node_tests.utils.cluster_nodes.ClusterKind(*values)

Bases: StrEnum

LOCAL = 'local'

TESTNET = 'testnet'

class cardano_node_tests.utils.cluster_nodes.ClusterType(*args, **kwargs)

Bases: Protocol

Protocol for cluster types.

NODES: ClassVar[frozenset[str]]

cluster_scripts: ScriptsTypes

create_addrs_data(*, cluster_obj: ClusterLib, destination_dir: str | Path = '.') → dict[str, dict[str, Any]]

Create addresses and their keys for usage in tests.

get_cluster_obj(*, command_era: str = '') → ClusterLib

Return instance of ClusterLib (cluster_obj).

property is_local: bool

Check if the cluster runs on a local testnet.

property is_testnet: bool

Check if the cluster runs on a long-running public network (preview, mainnet, etc.).

property testnet_type: str

Return testnet type (preview, preprod, etc.).

Returns an empty string on local cluster and “unknown” when the testnet is not recognized.

type: ClusterKind

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: ClusterType

Local cluster type (full cardano mode).

NODES: ClassVar[frozenset[str]] = frozenset({'bft1', 'pool1', 'pool2', 'pool3'})

create_addrs_data(*, cluster_obj: ClusterLib, destination_dir: str | Path = '.') → dict[str, dict[str, Any]]

Create addresses and their keys for usage in tests.

get_cluster_obj(*, command_era: str = '') → ClusterLib

Return instance of ClusterLib (cluster_obj).

property testnet_type: str

Return empty string, local cluster is not a testnet.

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: int | None, uptime: str | None, message: str = '')

Bases: object

message: str = ''

name: str

pid: int | None

status: str

uptime: str | None

class cardano_node_tests.utils.cluster_nodes.TestnetCluster

Bases: ClusterType

Testnet cluster type (full cardano mode).

NODES: ClassVar[frozenset[str]] = frozenset({'relay1'})

TESTNETS: ClassVar[dict[int, dict]] = {1506203091: {'shelley_start': '2020-07-29T21:44:51Z', 'type': Testnets.mainnet}, 1654041600: {'byron_epochs': 4, 'type': Testnets.preprod}, 1666656000: {'byron_epochs': 0, 'type': Testnets.preview}}

create_addrs_data(*, cluster_obj: ClusterLib, destination_dir: str | Path = '.') → dict[str, dict[str, Any]]

Create addresses and their keys for usage in tests.

get_cluster_obj(*, command_era: str = '') → ClusterLib

Return instance of ClusterLib (cluster_obj).

property testnet_type: str

Return testnet type (preview, preprod, 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(*values)

Bases: StrEnum

mainnet = 'mainnet'

preprod = 'preprod'

preview = 'preview'

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

Return path to socket file in the given cluster instance.

cardano_node_tests.utils.cluster_nodes.get_cluster_env() → ClusterEnv

Get cardano cluster environment.

cardano_node_tests.utils.cluster_nodes.get_cluster_type() → 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: ClusterLib) → dict

Load data for pools existing in the cluster environment.

cardano_node_tests.utils.cluster_nodes.reload_supervisor_config(*, instance_num: int | None = None, delay: int = 60) → None

Reload supervisor configuration.

cardano_node_tests.utils.cluster_nodes.restart_all_nodes(*, instance_num: int | None = None, delay: int = 60) → None

Restart all Cardano nodes of the running cluster.

cardano_node_tests.utils.cluster_nodes.restart_nodes(node_names: list[str], *, instance_num: int | None = None, delay: int = 60) → None

Restart list of Cardano nodes of the running cluster.

cardano_node_tests.utils.cluster_nodes.run_supervisorctl(args: list[str], *, instance_num: int | None = None, ignore_fail: bool = False) → bytes

Run supervisorctl command.

cardano_node_tests.utils.cluster_nodes.services_action(service_names: list[str], *, action: str, instance_num: int | None = None) → None

Perform action on services on the running cluster.

cardano_node_tests.utils.cluster_nodes.services_status(service_names: list[str] | None = None, *, instance_num: int | None = None) → list[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: ClusterLib, destination_dir: str | Path = '.') → Path

Set addresses and their keys up for usage in tests.

cardano_node_tests.utils.cluster_nodes.start_cluster(cmd: str, args: list[str]) → ClusterLib

Start cluster.

cardano_node_tests.utils.cluster_nodes.start_nodes(node_names: list[str], *, instance_num: int | None = 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: int | None = 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 altered by tests

• Setup of scripts and their configuration for starting of multiple cluster instances

class cardano_node_tests.utils.cluster_scripts.CustomCardonnayScripts(num_pools: int, scripts_dir: Path, ports_base: int)

Bases: LocalScripts

Scripts for starting local cluster.

Re-definition of Cardonnay Local Scripts that allows randomly selected localhost addresses.

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

Bases: ScriptsTypes

Scripts for starting local cluster.

copy_scripts_files(*, destdir: str | Path) → StartupFiles

Make copy of cluster scripts files located in this repository.

gen_split_topology_files(*, destdir: str | Path, instance_num: int, offset: int = 0) → None

Generate topology files for split network.

get_instance_ports(*, instance_num: int) → InstancePorts

Return ports mapping for given cluster instance.

prepare_scripts_files(*, destdir: str | Path, instance_num: int, scriptsdir: str | Path | None = None) → InstanceFiles

Prepare scripts files for starting and stopping cluster instance.

class cardano_node_tests.utils.cluster_scripts.ScriptsTypes(*args, **kwargs)

Bases: Protocol

Protocol for scripts of a given cluster instance type.

copy_scripts_files(*, destdir: str | Path) → StartupFiles

Make copy of cluster scripts files.

Testnet files and scripts can be copied and modified by tests before using them. E.g. we might want to change KES period, pool cost, etc.

gen_split_topology_files(*, destdir: str | Path, instance_num: int, offset: int = 0) → None

Generate topology files for split network.

get_instance_ports(*, instance_num: int) → InstancePorts

Return ports mapping for given cluster instance.

prepare_scripts_files(*, destdir: str | Path, instance_num: int, scriptsdir: str | Path | None = None) → 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: Path

start_script: Path

class cardano_node_tests.utils.cluster_scripts.TestnetScripts

Bases: 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', 'checkpoints*.json', '*snapshot*.json', 'dbsync-config.*', 'submit-api-config.*')

copy_scripts_files(*, destdir: str | Path) → StartupFiles

Make copy of cluster scripts files located in this repository.

gen_split_topology_files(*, destdir: str | Path, instance_num: int, offset: int = 0) → None

Generate topology files for split network.

Not supported on testnet, always raises NotImplementedError.

get_instance_ports(*, instance_num: int) → InstancePorts

Return ports mapping for given cluster instance.

prepare_scripts_files(*, destdir: str | Path, instance_num: int, scriptsdir: str | Path | None = None) → 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.cluster_scripts.get_testnet_dirs(base: Path) → set[str]

cardano_node_tests.utils.cluster_scripts.get_testnet_variant_scriptdir(*, testnet_variant: str) → Path | None

Get path to testnet variant scripts directory.

cardano_node_tests.utils.cluster_scripts.get_testnet_variant_scriptdir2(*, testnet_variant: str) → Path

Get path to testnet variant scripts directory.

Fails if the directory is not found.

cardano_node_tests.utils.cluster_scripts.get_testnet_variants() → list[str]

Get list of testnet variants.

cardano_node_tests.utils.clusterlib_utils module

Utilities that extends the functionality of cardano-clusterlib.

class cardano_node_tests.utils.clusterlib_utils.BuildMethods(*values)

Bases: StrEnum

BUILD = 'build'

BUILD_EST = 'build_estimate'

BUILD_RAW = 'build_raw'

class cardano_node_tests.utils.clusterlib_utils.ChainAccount(reserves: int, treasury: int)

Bases: object

reserves: int

treasury: int

class cardano_node_tests.utils.clusterlib_utils.KeyGenMethods(*values)

Bases: StrEnum

DIRECT = 'direct'

MNEMONIC = 'mnemonic'

class cardano_node_tests.utils.clusterlib_utils.NativeTokenRec(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[AddressRecord]

script: Path

token: str

token_mint_addr: AddressRecord

class cardano_node_tests.utils.clusterlib_utils.Token(coin: str, amount: int)

Bases: object

amount: int

coin: str

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: Callable | None = None)

Bases: object

arg: str

check_func: Callable | None = None

name: str = ''

value: Any

cardano_node_tests.utils.clusterlib_utils.build_and_submit_tx(*, cluster_obj: ClusterLib, name_template: str, src_address: str, submit_method: str = '', build_method: str = '', txins: list[UTXOData] | tuple[()] = (), txouts: list[TxOut] | tuple[()] = (), readonly_reference_txins: list[UTXOData] | tuple[()] = (), script_txins: list[ScriptTxIn] | tuple[()] = (), return_collateral_txouts: list[TxOut] | tuple[()] = (), total_collateral_amount: int | None = None, mint: list[Mint] | tuple[()] = (), tx_files: TxFiles | None = None, complex_certs: list[ComplexCert] | tuple[()] = (), complex_proposals: list[ComplexProposal] | tuple[()] = (), change_address: str = '', fee_buffer: int | None = None, raw_fee: int | None = None, required_signers: list[str | Path] | list[str] | list[Path] | tuple[()] = (), required_signer_hashes: list[str] | None = None, withdrawals: list[TxOut] | tuple[()] = (), script_withdrawals: list[ScriptWithdrawal] | tuple[()] = (), script_votes: list[ScriptVote] | tuple[()] = (), deposit: int | None = None, current_treasury_value: int | None = None, treasury_donation: int | None = None, invalid_hereafter: int | None = None, invalid_before: int | None = None, witness_override: int | None = None, witness_count_add: int = 0, byron_witness_count: int = 0, reference_script_size: int = 0, script_valid: bool = True, calc_script_cost_file: str | Path | None = None, join_txouts: bool = True, destination_dir: str | Path = '.', cli_asset_balancing: bool | None = None) → TxRawOutput

Build and submit a transaction.

Use submit_method to switch between cardano-cli transaction submit and submit-api. Use build_method to select one of the cardano-cli transaction build* commands:

• build

• build_raw

• build_estimate

cardano_node_tests.utils.clusterlib_utils.build_stake_multisig_script(*, cluster_obj: ClusterLib, script_name: str, script_type_arg: str, stake_vkey_files: Iterable[Path], required: int = 0, slot: int = 0, slot_type_arg: str = '') → Path

Build a stake multi-signature script.

Parameters:

• cluster_obj – An instance of clusterlib.ClusterLib.

• script_name – A name of the script.

• script_type_arg – A script type, see MultiSigTypeArgs.

• stake_vkey_files – A list of paths to stake vkey files.

• required – A number of required keys for the “atLeast” script type (optional).

• slot – A slot that sets script validity, depending on value of slot_type_arg (optional).

• slot_type_arg – A slot validity type, see MultiSlotTypeArgs (optional).

Returns:

A path to the script file.

Return type:

Path

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

Check that actual pool state corresponds with pool creation data.

cardano_node_tests.utils.clusterlib_utils.check_reference_script_policyid(name_template: str, cluster_obj: ClusterLib, script_file: str | Path, script_data: dict) → None

Check that the policy ID of a reference script file matches the original script policy ID.

cardano_node_tests.utils.clusterlib_utils.check_txins_spent(*, cluster_obj: ClusterLib, txins: list[UTXOData], wait_blocks: int = 2) → None

Check that txins were spent.

cardano_node_tests.utils.clusterlib_utils.check_updated_params(*, update_proposals: list[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_collaterals(*, cluster: ClusterLib, payment_addr: AddressRecord, temp_template: str, tx_outs: list[TxOut]) → list[UTXOData]

Create collateral UTxOs as required.

cardano_node_tests.utils.clusterlib_utils.create_payment_addr_records(*names: str, cluster_obj: ClusterLib, stake_vkey_file: str | Path | None = None, key_gen_method: KeyGenMethods = KeyGenMethods.DIRECT, destination_dir: str | Path = '.') → list[AddressRecord]

Create new payment address(es).

cardano_node_tests.utils.clusterlib_utils.create_pool_users(*, cluster_obj: ClusterLib, name_template: str, no_of_addr: int = 1, payment_key_gen_method: KeyGenMethods = KeyGenMethods.DIRECT, destination_dir: str | Path = '.') → list[PoolUser]

Create PoolUsers.

cardano_node_tests.utils.clusterlib_utils.create_reference_utxo(*, temp_template: str, cluster_obj: ClusterLib, payment_addr: AddressRecord, dst_addr: AddressRecord, script_file: Path, amount: int) → tuple[UTXOData, TxRawOutput]

Create a reference script UTxO.

cardano_node_tests.utils.clusterlib_utils.create_stake_addr_records(*names: str, cluster_obj: ClusterLib, destination_dir: str | Path = '.') → list[AddressRecord]

Create new stake address(es).

cardano_node_tests.utils.clusterlib_utils.datum_hash_from_txout(*, cluster_obj: ClusterLib, txout: TxOut) → str

Return datum hash from clusterlib.TxOut.

cardano_node_tests.utils.clusterlib_utils.deregister_stake_address(*, cluster_obj: ClusterLib, pool_user: PoolUser, name_template: str, deposit_amt: int = -1) → TxRawOutput | None

Deregister stake address.

cardano_node_tests.utils.clusterlib_utils.gen_byron_addr(*, cluster_obj: ClusterLib, name_template: str, destination_dir: str | Path = '.') → AddressRecord

Generate a Byron address and keys.

cardano_node_tests.utils.clusterlib_utils.gen_payment_addr_and_keys_from_mnemonic(*, name: str, cluster_obj: ClusterLib, size: Literal[12, 15, 18, 21, 24] = 24, stake_vkey_file: str | Path | None = None, stake_script_file: str | Path | None = None, destination_dir: str | Path = '.') → AddressRecord

Generate payment address and key pair.

Parameters:

• cluster_obj – An instance of clusterlib.ClusterLib.

• size – Number of words in the mnemonic (12, 15, 18, 21, or 24).

• name – A name of the address and key pair.

• stake_vkey_file – A path to corresponding stake vkey file (optional).

• stake_script_file – A path to corresponding payment script file (optional).

• destination_dir – A path to directory for storing artifacts (optional).

Returns:

A data container containing the address and

key pair / script file.

Return type:

clusterlib.AddressRecord

cardano_node_tests.utils.clusterlib_utils.get_blocks_before(*, cluster_obj: ClusterLib) → dict[str, int]

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

cardano_node_tests.utils.clusterlib_utils.get_chain_account_state(*, ledger_state: dict) → ChainAccount

Get chain account state from ledger state dict.

cardano_node_tests.utils.clusterlib_utils.get_delegation_state(*, cluster_obj: ClusterLib) → dict

Get delegationState section of ledger state.

cardano_node_tests.utils.clusterlib_utils.get_just_lovelace_utxos(*, address_utxos: list[UTXOData]) → list[UTXOData]

Get UTxOs with just Lovelace.

cardano_node_tests.utils.clusterlib_utils.get_ledger_state(*, cluster_obj: ClusterLib) → dict

Return the current ledger state info.

cardano_node_tests.utils.clusterlib_utils.get_plutus_b64(*, script_file: str | Path) → str

Get base64 encoded binary version of Plutus script from file.

cardano_node_tests.utils.clusterlib_utils.get_pparams_update_args(*, update_proposals: list[UpdateProposal]) → list[str]

Get cli arguments for pparams update action.

cardano_node_tests.utils.clusterlib_utils.get_protocol_version(cluster_obj: ClusterLib) → int

Get the protocol version major number.

cardano_node_tests.utils.clusterlib_utils.get_reference_script_size(*, script_file: str | Path) → int

Get the size of a script as it is accounted for when used as a reference script.

Since Conway, the ledger charges minFeeRefScriptCostPerByte for every byte of every reference script a transaction pulls in, be it through a reference input or through a spent UTxO that holds the script. The size is the size of the script as serialized by the ledger, which is the bare script for Plutus scripts and the CBOR encoded structure for simple scripts.

Parameters:

script_file – A path to the script file.

Returns:

A size of the reference script in bytes.

Return type:

int

cardano_node_tests.utils.clusterlib_utils.get_script_data_policyid(*, cluster_obj: ClusterLib, script_name: str, script_data: dict) → str

Get hash of the script data.

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, int | list]

Get uniform record for ledger state snapshot.

cardano_node_tests.utils.clusterlib_utils.get_stability_window(*, cluster_obj: ClusterLib) → int

Return the stability window (3k/f) in slots.

Rounded down, so that the result is never above the window the ledger itself uses. A caller that uses it as a limit therefore stays inside the window. The division is exact on every testnet variant, so the rounding only guards against a variant that picks a securityParam and activeSlotsCoeff that don’t divide.

Parameters:

cluster_obj – An instance of clusterlib.ClusterLib.

Returns:

The number of slots in the stability window.

Return type:

int

cardano_node_tests.utils.clusterlib_utils.get_stake_rec(*, stake_snapshot: dict) → dict

Get uniform record for stake key snapshot.

cardano_node_tests.utils.clusterlib_utils.get_utxo_ix_offset(*, utxos: list[UTXOData], txouts: list[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: Path) → Any

Load metadata from file containing transaction body.

cardano_node_tests.utils.clusterlib_utils.load_registered_pool_data(*, cluster_obj: ClusterLib, pool_name: str, pool_id: str) → PoolData

Load data of existing registered pool.

cardano_node_tests.utils.clusterlib_utils.load_tx_metadata(*, tx_body_file: Path) → TxMetadata

Load transaction metadata from file containing transaction body.

cardano_node_tests.utils.clusterlib_utils.mint_or_burn_sign(*, cluster_obj: ClusterLib, new_tokens: list[NativeTokenRec], temp_template: str, submit_method: str = SubmitMethods.CLI, build_method: str = BuildMethods.BUILD_RAW, sign_incrementally: bool = False) → 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: ClusterLib, new_tokens: list[NativeTokenRec], temp_template: str, invalid_hereafter: int | None = None, invalid_before: int | None = None, submit_method: str = SubmitMethods.CLI, build_method: str = BuildMethods.BUILD_RAW, sign_incrementally: bool = False) → 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: ClusterLib, temp_template: str, token_mint_addr: AddressRecord, issuer_addr: AddressRecord, amount: int) → list[NativeTokenRec]

Mint new token, sign using skeys.

cardano_node_tests.utils.clusterlib_utils.register_stake_address(*, cluster_obj: ClusterLib, pool_user: PoolUser, name_template: str, deposit_amt: int = -1) → TxRawOutput

Register stake address.

cardano_node_tests.utils.clusterlib_utils.save_ledger_state(*, cluster_obj: ClusterLib, state_name: str, ledger_state: dict | None = None, destination_dir: str | Path = '.') → 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.wait_for_epoch_interval(*, cluster_obj: 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: ClusterLib) → None

Wait until 4th epoch, if necessary, for first reward distribution.

cardano_node_tests.utils.clusterlib_utils.withdraw_reward_w_build(*, cluster_obj: ClusterLib, stake_addr_record: AddressRecord, dst_addr_record: AddressRecord, tx_name: str, verify: bool = True, destination_dir: str | Path = '.') → 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.custom_clusterlib module

Custom ClusterLib extended with functionality that is useful for testing.

class cardano_node_tests.utils.custom_clusterlib.ClusterLib(state_dir: str | Path, slots_offset: int | None = None, socket_path: str | Path = '', command_era: str = 'latest')

Bases: ClusterLib

cli(cli_args: list[str], timeout: float | None = None, add_default_args: bool = True) → CLIOut

Run the cardano-cli command.

Parameters:

• cli_args – A list of arguments for cardano-cli.

• timeout – A timeout for the command, in seconds (optional).

• add_default_args – Whether to add default arguments to the command (optional).

Returns:

A data container containing command stdout and stderr.

Return type:

clusterlib.CLIOut

property g_transaction: TransactionGroup

Transaction group.

class cardano_node_tests.utils.custom_clusterlib.TransactionGroup(clusterlib_obj: itp.ClusterLib)

Bases: TransactionGroup

submit_tx(tx_file: str | Path, txins: list[UTXOData], wait_blocks: int | None = None) → str

Create a .submitted status file when the Tx was successfully submitted.

cardano_node_tests.utils.custom_clusterlib.create_submitted_file(*, tx_file: str | Path) → None

Create a .submitted status file when the Tx was successfully submitted.

cardano_node_tests.utils.custom_clusterlib.record_cli_coverage(*, cli_args: list[str], coverage_dict: dict) → None

Record coverage info for CLI commands.

Parameters:

• cli_args – A list of command and it’s arguments.

• coverage_dict – A dictionary with coverage info.

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: ClusterLib, tx_raw_output: TxRawOutput, response: TxRecord) → None

Check a transaction in db-sync.

cardano_node_tests.utils.dbsync_check_tx.check_tx_collaterals(*, cluster_obj: ClusterLib, tx_raw_output: TxRawOutput, response: 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: ClusterLib, tx_raw_output: TxRawOutput, response: 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: TxRawOutput, response: 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: TxRawOutput, response: 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: TxRawOutput, response: 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: ClusterLib, tx_raw_output: TxRawOutput, response: 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: TxRawOutput, response: 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: ClusterLib, tx_raw_output: TxRawOutput, response: 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: TxRawOutput, response: 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: ClusterLib, tx_raw_output: TxRawOutput, response: 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: TxRawOutput, response: 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: TxRawOutput, response: TxRecord) → None

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

cardano_node_tests.utils.dbsync_check_tx.utxodata2txout(*, utxodata: UTxORecord | UTXOData) → TxOut

Convert UTxORecord or UTxOData to clusterlib.TxOut.

cardano_node_tests.utils.dbsync_check_tx.utxorecord2utxodata(utxorecord: UTxORecord) → 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, connection | None]] = {0: None}

cardano_node_tests.utils.dbsync_conn.close_all() → None

cardano_node_tests.utils.dbsync_conn.conn() → connection

cardano_node_tests.utils.dbsync_conn.reconn() → 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, reserves: Decimal, rewards: Decimal, utxo: Decimal, deposits_stake: Decimal, deposits_drep: Decimal, deposits_proposal: Decimal, fees: Decimal, block_id: int)

Bases: object

block_id: int

deposits_drep: Decimal

deposits_proposal: Decimal

deposits_stake: Decimal

epoch_no: int

fees: Decimal

id: int

reserves: Decimal

rewards: Decimal

slot_no: int

treasury: Decimal

utxo: Decimal

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

Bases: object

addr_view: str

amount: Decimal

cert_index: int

id: int

tx_id: int

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

Bases: object

block_no: int | None

epoch_no: int | None

epoch_slot_no: int | None

id: int

pool_id: str | None

previous_id: int | None

proto_major: int | None

proto_minor: int | None

slot_no: int | None

tx_count: int | None

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

Bases: object

address: str

tx_hash: memoryview

tx_out_id: int

utxo_ix: int

value: 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.DelegationVoteDBRow(id: int, stake_address_hash_view: str, drep_hash_view: str)

Bases: object

drep_hash_view: str

id: int

stake_address_hash_view: str

class cardano_node_tests.utils.dbsync_queries.DrepDistributionDBRow(id: int, hash_id: int, amount: int, epoch_no: int, drep_hash_view: str)

Bases: object

amount: int

drep_hash_view: str

epoch_no: int

hash_id: int

id: int

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 | None, 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 | None

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.EpochParamDBRow(id: int, epoch_no: int, 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, protocol_major: int, protocol_minor: int, min_utxo_value: int, min_pool_cost: int, nonce: memoryview, 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, block_id: int, extra_entropy: memoryview | None, coins_per_utxo_size: int, pvt_motion_no_confidence: float, pvt_committee_normal: float, pvt_committee_no_confidence: float, pvt_hard_fork_initiation: float, dvt_motion_no_confidence: float, dvt_committee_normal: float, dvt_committee_no_confidence: float, dvt_update_to_constitution: float, dvt_hard_fork_initiation: float, dvt_p_p_network_group: float, dvt_p_p_economic_group: float, dvt_p_p_technical_group: float, dvt_p_p_gov_group: float, dvt_treasury_withdrawal: float, committee_min_size: int, committee_max_term_length: int, gov_action_lifetime: int, gov_action_deposit: int, drep_deposit: int, drep_activity: int, pvtpp_security_group: float, min_fee_ref_script_cost_per_byte: float)

Bases: object

block_id: int

coins_per_utxo_size: int

collateral_percent: int

committee_max_term_length: int

committee_min_size: int

cost_model_id: int

decentralisation: float

drep_activity: int

drep_deposit: int

dvt_committee_no_confidence: float

dvt_committee_normal: float

dvt_hard_fork_initiation: float

dvt_motion_no_confidence: float

dvt_p_p_economic_group: float

dvt_p_p_gov_group: float

dvt_p_p_network_group: float

dvt_p_p_technical_group: float

dvt_treasury_withdrawal: float

dvt_update_to_constitution: float

epoch_no: int

extra_entropy: memoryview | None

gov_action_deposit: int

gov_action_lifetime: int

id: int

influence: float

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_fee_ref_script_cost_per_byte: float

min_pool_cost: int

min_utxo_value: int

monetary_expand_rate: float

nonce: memoryview

optimal_pool_count: int

pool_deposit: int

price_mem: float

price_step: float

protocol_major: int

protocol_minor: int

pvt_committee_no_confidence: float

pvt_committee_normal: float

pvt_hard_fork_initiation: float

pvt_motion_no_confidence: float

pvtpp_security_group: float

treasury_growth_rate: float

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.EpochStateDBRow(id: int, committee_id: int, no_confidence_id: int | None, constitution_id: int, epoch_no: int)

Bases: object

committee_id: int

constitution_id: int

epoch_no: int

id: int

no_confidence_id: int | None

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

Bases: object

action_ix: int

deposit: int

description: dict

dropped_epoch: int | None

enacted_epoch: int | None

expiration: int

expired_epoch: int | None

id: int

param_proposal: int | None

prev_gov_action_proposal: int | None

ratified_epoch: int | None

return_address: int

tx_id: int

type: str

voting_anchor_id: int

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

Bases: object

bytes: memoryview

id: int

json: Any

key: Decimal

tx_id: int

class cardano_node_tests.utils.dbsync_queries.NewCommitteeInfoDBRow(id: int, tx_id: int, action_ix: int, quorum_numerator: int, quorum_denominator: int)

Bases: object

action_ix: int

id: int

quorum_denominator: int

quorum_numerator: int

tx_id: int

class cardano_node_tests.utils.dbsync_queries.NewCommitteeMemberDBRow(id: int, committee_id: int, committee_hash: memoryview, expiration_epoch: int)

Bases: object

committee_hash: memoryview

committee_id: int

expiration_epoch: int

id: int

class cardano_node_tests.utils.dbsync_queries.NewConstitutionInfoDBRow(id: int, script_hash: memoryview | None, gov_action_type: str, gap_id: int, tx_id: int, action_ix: int)

Bases: object

action_ix: int

gap_id: int

gov_action_type: str

id: int

script_hash: memoryview | None

tx_id: int

class cardano_node_tests.utils.dbsync_queries.OffChainVoteDataDBRow(data_id: int, data_vot_anchor_id: int, data_hash: memoryview, data_json: dict, data_bytes: memoryview, data_warning: str | None, data_language: str, data_comment: str | None, data_is_valid: bool | None, auth_name: str | None, auth_wit_alg: str | None, auth_pub_key: str | None, auth_signature: str | None, auth_warning: str | None, ext_update_id: int | None, ext_update_title: str | None, ext_update_uri: str | None, gov_act_id: int | None, gov_act_title: str | None, gov_act_abstract: str | None, gov_act_motivation: str | None, gov_act_rationale: str | None, ref_id: int | None, ref_label: str | None, ref_uri: str | None, ref_hash_digest: str | None, ref_hash_alg: str | None, vot_anchor_url: str, vot_anchor_data_hash: memoryview, vot_anchor_type: str, vot_anchor_block_id: int)

Bases: object

auth_name: str | None

auth_pub_key: str | None

auth_signature: str | None

auth_warning: str | None

auth_wit_alg: str | None

data_bytes: memoryview

data_comment: str | None

data_hash: memoryview

data_id: int

data_is_valid: bool | None

data_json: dict

data_language: str

data_vot_anchor_id: int

data_warning: str | None

ext_update_id: int | None

ext_update_title: str | None

ext_update_uri: str | None

gov_act_abstract: str | None

gov_act_id: int | None

gov_act_motivation: str | None

gov_act_rationale: str | None

gov_act_title: str | None

ref_hash_alg: str | None

ref_hash_digest: str | None

ref_id: int | None

ref_label: str | None

ref_uri: str | None

vot_anchor_block_id: int

vot_anchor_data_hash: memoryview

vot_anchor_type: str

vot_anchor_url: str

class cardano_node_tests.utils.dbsync_queries.OffChainVoteDrepDataDBRow(id: int, hash: memoryview, language: str, comment: str | None, json: dict, bytes: memoryview, warning: str | None, is_valid: bool | None, payment_address: str, given_name: str, objectives: str, motivations: str, qualifications: str, image_url: str, image_hash: str | None)

Bases: object

bytes: memoryview

comment: str | None

given_name: str

hash: memoryview

id: int

image_hash: str | None

image_url: str

is_valid: bool | None

json: dict

language: str

motivations: str

objectives: str

payment_address: str

qualifications: str

warning: str | None

class cardano_node_tests.utils.dbsync_queries.OffChainVoteFetchErrorDBRow(id: int, voting_anchor_id: int, fetch_error: str)

Bases: object

fetch_error: str

id: int

voting_anchor_id: int

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

Bases: object

coins_per_utxo_size: int | None

collateral_percent: int | None

committee_max_term_length: int | None

committee_min_size: int | None

cost_model_id: int | None

decentralisation: float | None

drep_activity: Decimal | None

drep_deposit: int | None

dvt_committee_no_confidence: float | None

dvt_committee_normal: float | None

dvt_hard_fork_initiation: float | None

dvt_motion_no_confidence: float | None

dvt_p_p_economic_group: float | None

dvt_p_p_gov_group: float | None

dvt_p_p_network_group: float | None

dvt_p_p_technical_group: float | None

dvt_treasury_withdrawal: float | None

dvt_update_to_constitution: float | None

entropy: memoryview | None

epoch_no: int | None

gov_action_deposit: int | None

gov_action_lifetime: int | None

id: int

influence: float | None

key: memoryview | None

key_deposit: int | None

max_bh_size: int | None

max_block_ex_mem: int | None

max_block_ex_steps: int | None

max_block_size: int | None

max_collateral_inputs: int | None

max_epoch: int | None

max_tx_ex_mem: int | None

max_tx_ex_steps: int | None

max_tx_size: int | None

max_val_size: int | None

min_fee_a: int | None

min_fee_b: int | None

min_fee_ref_script_cost_per_byte: float | None

min_pool_cost: int | None

min_utxo_value: int | None

monetary_expand_rate: float | None

optimal_pool_count: int | None

pool_deposit: int | None

price_mem: float | None

price_step: float | None

protocol_major: int | None

protocol_minor: int | None

pvt_committee_no_confidence: float | None

pvt_committee_normal: float | None

pvt_hard_fork_initiation: float | None

pvt_motion_no_confidence: float | None

pvtpp_security_group: float | None

registered_tx_id: int | None

treasury_growth_rate: float | None

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 | None, margin: Decimal, fixed_cost: int, deposit: Decimal | None, registered_tx_id: int, metadata_url: str | None, metadata_hash: memoryview | None, owner_stake_address_id: int, owner: memoryview, ipv4: str | None, ipv6: str | None, dns_name: str | None, port: int | None, retire_cert_index: int | None, retire_announced_tx_id: int | None, retiring_epoch: int | None)

Bases: object

active_epoch_no: int

cert_index: int

deposit: Decimal | None

dns_name: str | None

fixed_cost: int

hash: memoryview

id: int

ipv4: str | None

ipv6: str | None

margin: Decimal

meta_id: int | None

metadata_hash: memoryview | None

metadata_url: str | None

owner: memoryview

owner_stake_address_id: int

pledge: int

port: int | None

registered_tx_id: int

retire_announced_tx_id: int | None

retire_cert_index: int | None

retiring_epoch: int | None

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, reserves: Decimal, tx_id: int)

Bases: object

cert_index: int

id: int

reserves: Decimal

treasury: 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, earned_epoch: int, spendable_epoch: int, pool_id: str | None = '')

Bases: object

address: str

amount: Decimal

earned_epoch: int

pool_id: str | None = ''

spendable_epoch: int

type: str

class cardano_node_tests.utils.dbsync_queries.SchemaVersion

Bases: object

Query and cache db-sync schema version.

classmethod stages() → 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: int | None)

Bases: object

hash: memoryview

id: int

serialised_size: int | None

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: int | None, pool_id: str | None, address: str | None)

Bases: object

active_epoch_no: int | None

address: str | None

pool_id: str | None

tx_id: int

class cardano_node_tests.utils.dbsync_queries.TreasuryWithdrawalDBRow(id: int, tx_id: int, action_ix: int, expiration: int, ratified_epoch: int | None, enacted_epoch: int | None, dropped_epoch: int | None, expired_epoch: int | None, addr_view: str, amount: Decimal)

Bases: object

action_ix: int

addr_view: str

amount: Decimal

dropped_epoch: int | None

enacted_epoch: int | None

expiration: int

expired_epoch: int | None

id: int

ratified_epoch: int | None

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, fee: Decimal, deposit: int, size: int, invalid_before: Decimal | None, invalid_hereafter: Decimal | None, treasury_donation: int, tx_out_id: int | None, tx_out_tx_id: int | None, utxo_ix: int | None, tx_out_addr: str | None, tx_out_addr_has_script: bool | None, tx_out_value: Decimal | None, tx_out_data_hash: memoryview | None, tx_out_inline_datum_hash: memoryview | None, tx_out_reference_script_hash: memoryview | None, 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: int | None, ma_tx_out_policy: memoryview | None, ma_tx_out_name: memoryview | None, ma_tx_out_quantity: Decimal | None, ma_tx_mint_id: int | None, ma_tx_mint_policy: memoryview | None, ma_tx_mint_name: memoryview | None, ma_tx_mint_quantity: Decimal | None)

Bases: object

block_id: int

block_index: int

collateral_count: int

collateral_out_count: int

deposit: int

extra_key_witness_count: int

fee: Decimal

invalid_before: Decimal | None

invalid_hereafter: Decimal | None

ma_tx_mint_id: int | None

ma_tx_mint_name: memoryview | None

ma_tx_mint_policy: memoryview | None

ma_tx_mint_quantity: Decimal | None

ma_tx_out_id: int | None

ma_tx_out_name: memoryview | None

ma_tx_out_policy: memoryview | None

ma_tx_out_quantity: Decimal | None

metadata_count: int

out_sum: 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

treasury_donation: int

tx_hash: memoryview

tx_id: int

tx_out_addr: str | None

tx_out_addr_has_script: bool | None

tx_out_data_hash: memoryview | None

tx_out_id: int | None

tx_out_inline_datum_hash: memoryview | None

tx_out_reference_script_hash: memoryview | None

tx_out_tx_id: int | None

tx_out_value: Decimal | None

utxo_ix: int | None

withdrawal_count: int

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

Bases: object

address: str

ma_tx_out_id: int | None

ma_tx_out_name: memoryview | None

ma_tx_out_policy: memoryview | None

ma_tx_out_quantity: Decimal | None

reference_script_bytes: memoryview | None

reference_script_hash: memoryview | None

reference_script_json: dict | None

reference_script_type: str | None

tx_hash: memoryview

tx_out_id: int

utxo_ix: int

value: Decimal

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

Bases: object

address: str

reference_script_bytes: memoryview | None

reference_script_hash: memoryview | None

reference_script_json: dict | None

reference_script_type: str | None

tx_hash: memoryview

tx_out_id: int

utxo_ix: int

value: Decimal

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

Bases: object

data_hash: memoryview | None

has_script: bool

payment_address: str

stake_address: str | None

tx_hash: memoryview

utxo_ix: int

value: int

class cardano_node_tests.utils.dbsync_queries.VotingProcedureDBRow(id: int, voter_role: str, committee_voter: int | None, drep_voter: int | None, pool_voter: int | None, vote: str)

Bases: object

committee_voter: int | None

drep_voter: int | None

id: int

pool_voter: int | None

vote: str

voter_role: str

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.db_transaction() → Iterator[None]

Transaction manager that works with connection pool.

cardano_node_tests.utils.dbsync_queries.delete_reserved_pool_tickers() → int

Delete all records from reserved_pool_ticker and return the number of affected rows.

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

cardano_node_tests.utils.dbsync_queries.query_ada_pots(*, epoch_from: int = 0, epoch_to: int = 99999999) → Generator[ADAPotsDBRow, None, None]

Query ADA pots record in db-sync.

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

Query reward records for stake address in db-sync.

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

Query instant 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[BlockDBRow, None, None]

Query block records in db-sync.

cardano_node_tests.utils.dbsync_queries.query_collateral_tx_ins(*, txhash: str) → Generator[TxInNoMADBRow, None, None]

Query transaction collateral txins in db-sync.

cardano_node_tests.utils.dbsync_queries.query_collateral_tx_outs(*, txhash: str) → Generator[CollateralTxOutDBRow, None, None]

Query transaction collateral txouts in db-sync.

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

Query committee registration in db-sync.

cardano_node_tests.utils.dbsync_queries.query_committee_members(*, committee_id: int) → Generator[NewCommitteeMemberDBRow, None, None]

Query committee members in db-sync.

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

Query committee registration in db-sync.

cardano_node_tests.utils.dbsync_queries.query_cost_model(*, model_id: int = -1, epoch_no: int = -1) → dict[str, dict[str, Any]]

Query cost model record in db-sync.

If model_id is specified, query the cost model that corresponds to the given id. If epoch_no is specified, query the cost model used in the given epoch. Otherwise query the latest cost model.

cardano_node_tests.utils.dbsync_queries.query_datum(*, datum_hash: str) → Generator[DatumDBRow, None, None]

Query datum record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_db_size() → int

Query database size in MB in db-sync.

cardano_node_tests.utils.dbsync_queries.query_db_sync_progress() → float | None

Calculate blockchain sync percentage (0-100).

Returns:

Sync percentage (0-100), or None when there are no blocks in the database.

cardano_node_tests.utils.dbsync_queries.query_delegation_vote(*, txhash: str) → Generator[DelegationVoteDBRow, None, None]

Query delegation_vote table in db-sync.

cardano_node_tests.utils.dbsync_queries.query_drep_distr(*, drep_hash: str, epoch_no: int) → Generator[DrepDistributionDBRow, None, None]

Query drep voting power in db-sync.

cardano_node_tests.utils.dbsync_queries.query_drep_registration(*, drep_hash: str, drep_deposit: int = 2000000) → Generator[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[EpochDBRow, None, None]

Query epoch records in db-sync.

cardano_node_tests.utils.dbsync_queries.query_epoch_current(*, epoch_from: int = 0, epoch_to: int = 99999999) → Generator[EpochDBRow, None, None]

Query the live epoch_current view in db-sync (since db-sync 13.7.2).

Unlike the public epoch view (finalized rows UNION the live aggregation), this reads the in-progress current-epoch branch directly. It is empty unless live epoch sync is enabled (see query_epoch_sync_enabled).

cardano_node_tests.utils.dbsync_queries.query_epoch_param(*, epoch_no: int = 0) → EpochParamDBRow

Query epoch param record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_epoch_stake(*, pool_id_bech32: str, epoch_number: int) → Generator[EpochStakeDBRow, None, None]

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

cardano_node_tests.utils.dbsync_queries.query_epoch_state(*, epoch_no: int) → Generator[EpochStateDBRow, None, None]

Query epoch_state table in db-sync.

cardano_node_tests.utils.dbsync_queries.query_epoch_sync_enabled() → bool

Query whether live epoch sync is enabled in db-sync (since db-sync 13.7.2).

When disabled, both the epoch_current and epoch views return no rows by design.

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[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_new_committee_info(*, txhash: str) → Generator[NewCommitteeInfoDBRow, None, None]

Query new committee proposed in db-sync.

cardano_node_tests.utils.dbsync_queries.query_new_constitution(*, txhash: str) → Generator[NewConstitutionInfoDBRow, None, None]

Query new constitution proposed in db-sync.

cardano_node_tests.utils.dbsync_queries.query_off_chain_pool_data(*, pool_id_bech32: str) → Generator[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[PoolOffChainFetchErrorDBRow, None, None]

Query Off_Chain_Pool_Fetch_Error record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_off_chain_vote_data(*, data_hash: str) → Generator[OffChainVoteDataDBRow, None, None]

Query the off chain vote data in db-sync.

cardano_node_tests.utils.dbsync_queries.query_off_chain_vote_drep_data(*, voting_anchor_id: int) → Generator[OffChainVoteDrepDataDBRow, None, None]

Query off_chain_vote_drep_data table in db-sync.

cardano_node_tests.utils.dbsync_queries.query_off_chain_vote_fetch_error(*, voting_anchor_id: int) → Generator[OffChainVoteFetchErrorDBRow, None, None]

Query off_chain_vote_fetch_error table in db-sync.

cardano_node_tests.utils.dbsync_queries.query_param_proposal(*, txhash: str = '') → 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[PoolDataDBRow, None, None]

Query pool data record in db-sync.

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

Query transaction redeemers in db-sync.

cardano_node_tests.utils.dbsync_queries.query_reference_tx_ins(*, txhash: str) → Generator[TxInNoMADBRow, None, None]

Query transaction reference txins in db-sync.

cardano_node_tests.utils.dbsync_queries.query_rows_count(*, table: str, column: str | None = None, condition: str | None = None, lock: bool = False, lock_timeout: str = '5s') → int

Query row count with optional table locking for atomic snapshots.

Parameters:

• table – Table name

• column – Column name (optional)

• condition – SQL condition (required if column is provided)

• lock – Whether to lock table during query

• lock_timeout – PostgreSQL timeout string (e.g., ‘5s’, ‘1min’)

Returns:

Count of matching rows (0 if none match or table is empty)

Raises:

• ValueError – For invalid inputs

• RuntimeError – If lock times out

cardano_node_tests.utils.dbsync_queries.query_scripts(*, txhash: str) → Generator[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_treasury_withdrawal(*, txhash: str) → Generator[TreasuryWithdrawalDBRow, None, None]

Query treasury_withdrawal table in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx(*, txhash: str) → Generator[TxDBRow, None, None]

Query a transaction in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_ins(*, txhash: str) → Generator[TxInDBRow, None, None]

Query transaction txins in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_metadata(*, txhash: str) → Generator[MetadataDBRow, None, None]

Query transaction metadata in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_pot_transfers(*, txhash: str) → Generator[PotTransferDBRow, None, None]

Query transaction MIR certificate records in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_reserve(*, txhash: str) → Generator[ADAStashDBRow, None, None]

Query transaction reserve record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_stake_deleg(*, txhash: str) → Generator[StakeDelegDBRow, None, None]

Query stake registration record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_stake_dereg(*, txhash: str) → Generator[StakeAddrDBRow, None, None]

Query stake deregistration record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_stake_reg(*, txhash: str) → Generator[StakeAddrDBRow, None, None]

Query stake registration record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_treasury(*, txhash: str) → Generator[ADAStashDBRow, None, None]

Query transaction treasury record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_tx_withdrawal(*, txhash: str) → Generator[WithdrawalDBRow, None, None]

Query reward withdrawal record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_utxo(*, address: str) → Generator[UTxODBRow, None, None]

Query UTxOs for payment address in db-sync.

cardano_node_tests.utils.dbsync_queries.query_view_names() → list[str]

Query view names in db-sync.

cardano_node_tests.utils.dbsync_queries.query_voting_procedure(*, txhash: str) → Generator[VotingProcedureDBRow, None, None]

Query voting_procedure table in db-sync.

cardano_node_tests.utils.dbsync_service_manager module

Functionality for managing db-sync service.

class cardano_node_tests.utils.dbsync_service_manager.Column

Bases: object

class Redeemer(*values)

Bases: StrEnum

SCRIPT_HASH = 'redeemer.script_hash'

class Tx(*values)

Bases: StrEnum

FEE = 'tx.fee'

class cardano_node_tests.utils.dbsync_service_manager.DBSyncConfigBuilder

Bases: object

build() → dict[str, Any]

with_governance(*, value: SettingState) → Self

with_ledger(*, value: LedgerMode) → Self

with_metadata(*, enable: bool, keys: list[int] | None = None) → Self

with_multi_asset(*, enable: bool) → Self

with_offchain_pool_data(*, value: SettingState) → Self

with_plutus(*, enable: bool) → Self

with_pool_stat(*, value: SettingState) → Self

with_preset(*, preset: Preset) → Self

with_remove_jsonb_from_schema(*, value: SettingState) → Self

with_shelley(*, enable: bool) → Self

with_tx_cbor(*, value: SettingState) → Self

with_tx_out(*, value: TxOutMode, force_tx_in: bool | None = None, use_address_table: bool | None = None) → Self

class cardano_node_tests.utils.dbsync_service_manager.DBSyncManager

Bases: object

get_config_builder() → DBSyncConfigBuilder

Get a fresh DBSyncConfigBuilder config builder instance with default config.

is_db_sync_running() → bool

Check if the cardano-db-sync service is actively running.

Returns:

True if the service is running and executable, False otherwise.

Return type:

bool

recreate_database() → None

Reinitializes the PostgreSQL database by running the postgres-setup.sh script.

Warning

This is a DESTRUCTIVE operation! It will delete existing data and recreate the database from scratch.

restart_with_config(*, custom_config: dict | DBSyncConfigBuilder | None = None) → Path

Restart cardano-db-sync with a new configuration, ensuring atomic updates.

Steps:

1. Stops the service.

2. Recreates the database.

3. Applies the new config.

4. Restarts the service and waits for sync completion.

Parameters:

custom_config – Either a config dict or DBSyncConfigBuilder instance.

Returns:

Path to the updated config file.

start_db_sync() → None

Start cardano-db-sync service.

stop_db_sync() → None

Stop cardano-db-sync service.

update_config(*, config: dict | DBSyncConfigBuilder | None) → Path

Update the dbsync-config.yaml file with new settings.

Parameters:

config – Config dict or DBSyncConfigBuilder instance.

Returns:

Path to the modified config file.

class cardano_node_tests.utils.dbsync_service_manager.LedgerMode(*values)

Bases: StrEnum

DISABLE = 'disable'

ENABLE = 'enable'

IGNORE = 'ignore'

class cardano_node_tests.utils.dbsync_service_manager.MetadataConfig(enable: bool = True, keys: list[int] | None = None)

Bases: object

enable: bool = True

keys: list[int] | None = None

class cardano_node_tests.utils.dbsync_service_manager.MultiAssetConfig(enable: bool = True)

Bases: object

enable: bool = True

class cardano_node_tests.utils.dbsync_service_manager.PlutusConfig(enable: bool = True)

Bases: object

enable: bool = True

class cardano_node_tests.utils.dbsync_service_manager.Preset(*values)

Bases: StrEnum

DISABLE_ALL = 'disable_all'

FULL = 'full'

ONLY_GOVERNANCE = 'only_governance'

ONLY_UTXO = 'only_utxo'

class cardano_node_tests.utils.dbsync_service_manager.SettingState(*values)

Bases: StrEnum

DISABLE = 'disable'

ENABLE = 'enable'

class cardano_node_tests.utils.dbsync_service_manager.ShelleyConfig(enable: bool = True)

Bases: object

enable: bool = True

class cardano_node_tests.utils.dbsync_service_manager.Table(*values)

Bases: StrEnum

Enum representing all tables in the Cardano db-sync database.

ADA_POTS = 'ada_pots'

ADDRESS = 'address'

BLOCK = 'block'

COLLATERAL_TX_IN = 'collateral_tx_in'

COLLATERAL_TX_OUT = 'collateral_tx_out'

COMMITTEE = 'committee'

COMMITTEE_DE_REGISTRATION = 'committee_de_registration'

COMMITTEE_HASH = 'committee_hash'

COMMITTEE_MEMBER = 'committee_member'

COMMITTEE_REGISTRATION = 'committee_registration'

CONSTITUTION = 'constitution'

COST_MODEL = 'cost_model'

DATUM = 'datum'

DELEGATION = 'delegation'

DELEGATION_VOTE = 'delegation_vote'

DELISTED_POOL = 'delisted_pool'

DREP_DISTR = 'drep_distr'

DREP_HASH = 'drep_hash'

DREP_REGISTRATION = 'drep_registration'

EPOCH_FINALIZED = 'epoch_finalized'

EPOCH_PARAM = 'epoch_param'

EPOCH_STAKE = 'epoch_stake'

EPOCH_STAKE_PROGRESS = 'epoch_stake_progress'

EPOCH_STATE = 'epoch_state'

EPOCH_SYNC_ENABLED = 'epoch_sync_enabled'

EPOCH_SYNC_TIME = 'epoch_sync_time'

EVENT_INFO = 'event_info'

EXTRA_KEY_WITNESS = 'extra_key_witness'

EXTRA_MIGRATIONS = 'extra_migrations'

GOV_ACTION_PROPOSAL = 'gov_action_proposal'

MA_TX_MINT = 'ma_tx_mint'

MA_TX_OUT = 'ma_tx_out'

META = 'meta'

MULTI_ASSET = 'multi_asset'

NEW_COMMITTEE = 'new_committee'

OFF_CHAIN_POOL_DATA = 'off_chain_pool_data'

OFF_CHAIN_POOL_FETCH_ERROR = 'off_chain_pool_fetch_error'

OFF_CHAIN_VOTE_AUTHOR = 'off_chain_vote_author'

OFF_CHAIN_VOTE_DATA = 'off_chain_vote_data'

OFF_CHAIN_VOTE_DREP_DATA = 'off_chain_vote_drep_data'

OFF_CHAIN_VOTE_EXTERNAL_UPDATE = 'off_chain_vote_external_update'

OFF_CHAIN_VOTE_FETCH_ERROR = 'off_chain_vote_fetch_error'

OFF_CHAIN_VOTE_GOV_ACTION_DATA = 'off_chain_vote_gov_action_data'

OFF_CHAIN_VOTE_REFERENCE = 'off_chain_vote_reference'

PARAM_PROPOSAL = 'param_proposal'

POOL_HASH = 'pool_hash'

POOL_METADATA_REF = 'pool_metadata_ref'

POOL_OWNER = 'pool_owner'

POOL_RELAY = 'pool_relay'

POOL_RETIRE = 'pool_retire'

POOL_STAT = 'pool_stat'

POOL_UPDATE = 'pool_update'

POT_TRANSFER = 'pot_transfer'

REDEEMER = 'redeemer'

REDEEMER_DATA = 'redeemer_data'

REFERENCE_TX_IN = 'reference_tx_in'

RESERVE = 'reserve'

RESERVED_POOL_TICKER = 'reserved_pool_ticker'

REVERSE_INDEX = 'reverse_index'

REWARD = 'reward'

REWARD_REST = 'reward_rest'

SCHEMA_VERSION = 'schema_version'

SCRIPT = 'script'

SLOT_LEADER = 'slot_leader'

STAKE_ADDRESS = 'stake_address'

STAKE_DEREGISTRATION = 'stake_deregistration'

STAKE_REGISTRATION = 'stake_registration'

TREASURY = 'treasury'

TREASURY_WITHDRAWAL = 'treasury_withdrawal'

TX = 'tx'

TX_CBOR = 'tx_cbor'

TX_IN = 'tx_in'

TX_METADATA = 'tx_metadata'

TX_OUT = 'tx_out'

VOTING_ANCHOR = 'voting_anchor'

VOTING_PROCEDURE = 'voting_procedure'

class cardano_node_tests.utils.dbsync_service_manager.TxOutConfig(value: cardano_node_tests.utils.dbsync_service_manager.TxOutMode = <TxOutMode.ENABLE: 'enable'>, force_tx_in: bool | None = None, use_address_table: bool | None = None)

Bases: object

force_tx_in: bool | None = None

use_address_table: bool | None = None

value: TxOutMode = 'enable'

class cardano_node_tests.utils.dbsync_service_manager.TxOutMode(*values)

Bases: StrEnum

BOOTSTRAP = 'bootstrap'

CONSUMED = 'consumed'

DISABLE = 'disable'

ENABLE = 'enable'

PRUNE = 'prune'

class cardano_node_tests.utils.dbsync_service_manager.View(*values)

Bases: StrEnum

Enum representing the views in the Cardano db-sync database (since db-sync 13.7.2).

EPOCH = 'epoch'

EPOCH_CURRENT = 'epoch_current'

cardano_node_tests.utils.dbsync_snapshot_service module

class cardano_node_tests.utils.dbsync_snapshot_service.DBSyncSnapshotService

Bases: object

Service class to interact with DB-Sync S3 Snapshots repository.

BUCKET_URL: str = 'https://update-cardano-mainnet.iohk.io'

ROOT_PREFIX: str = 'cardano-db-sync/'

get_latest_snapshot(version: str) → SnapshotFile

Find the latest snapshot file for a given version.

get_latest_version() → str

Find the numerically latest db-sync version folder.

class cardano_node_tests.utils.dbsync_snapshot_service.SnapshotFile(key: str, name: str, last_modified: datetime, size: int)

Bases: object

Dataclass to hold parsed snapshot file information.

key: str

last_modified: datetime

name: str

size: int

size_gb: float

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 | None, 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 | None

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.OffChainVoteDataRecord(id: int, vot_anchor_id: int, hash: str, json: dict, bytes: str, warning: str | None, language: str, comment: str | None, is_valid: bool | None, authors: list[dict[str, Any]], references: list[dict[str, Any]], gov_action_data: dict[str, Any], external_updates: list[dict[str, Any]], voting_anchor: dict[str, Any])

Bases: object

authors: list[dict[str, Any]]

bytes: str

comment: str | None

external_updates: list[dict[str, Any]]

gov_action_data: dict[str, Any]

hash: str

id: int

is_valid: bool | None

json: dict

language: str

references: list[dict[str, Any]]

vot_anchor_id: int

voting_anchor: dict[str, Any]

warning: str | None

class cardano_node_tests.utils.dbsync_types.PaymentAddrRecord(payment_address: str, stake_address: str | None, amount_sum: int, utxos: list[cardano_node_tests.utils.dbsync_types.GetUTxORecord])

Bases: object

amount_sum: int

payment_address: str

stake_address: str | None

utxos: list[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 | None, 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 | None, retire_announced_tx_id: int | None, retiring_epoch: int | None)

Bases: object

active_epoch_no: int

cert_index: int

fixed_cost: int

hash: str

id: int

margin: float

meta_id: int | None

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 | None

retire_cert_index: int | None

retiring_epoch: int | None

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[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: TxDBRow

ma_utxo_out: list[UTxORecord]

mint_utxo_out: list[UTxORecord]

utxo_out: list[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: int | None, invalid_hereafter: int | None, treasury_donation: 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[UTXOData]

collaterals: list[UTxORecord]

deposit: int

extra_key_witness: list[str]

fee: int

invalid_before: int | None

invalid_hereafter: int | None

metadata: list[MetadataRecord]

mint: list[UTxORecord]

out_sum: int

pot_transfers: list[PotTransferRecord]

redeemers: list[RedeemerRecord]

reference_inputs: list[UTxORecord]

reserve: list[ADAStashRecord]

scripts: list[ScriptRecord]

size: int

stake_delegation: list[DelegationRecord]

stake_deregistration: list[str]

stake_registration: list[str]

treasury: list[ADAStashRecord]

treasury_donation: int

tx_hash: str

tx_id: int

txins: list[UTxORecord]

txouts: list[UTxORecord]

withdrawals: list[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: str | dict | None = None, reference_script: dict | None = None, reference_script_hash: str = '')

Bases: object

address: str

amount: int

coin: str = 'lovelace'

datum_hash: str = ''

decoded_coin: str = ''

inline_datum: str | dict | None = None

inline_datum_hash: str = ''

reference_script: dict | None = None

reference_script_hash: str = ''

utxo_hash: str

utxo_ix: int

cardano_node_tests.utils.dbsync_utils module

Functionality for interacting with db-sync.

class cardano_node_tests.utils.dbsync_utils.ActionTypes(*values)

Bases: StrEnum

COMMITTEE = 'committee'

CONSTITUTION = 'constitution'

exception cardano_node_tests.utils.dbsync_utils.DbSyncNoResponseError(detail: str)

Bases: Exception

Raised when no response is returned from db-sync.

exception cardano_node_tests.utils.dbsync_utils.DbSyncTimeoutError

Bases: TimeoutError

Raised when data does not appear in db-sync within a timeout.

Subclasses TimeoutError so existing callers that catch TimeoutError keep working, while letting new code catch db-sync waiting timeouts specifically (a low-level socket.timeout is a TimeoutError too since Python 3.10). Note that TimeoutError subclasses OSError, so a broad except OSError would absorb this error as well.

cardano_node_tests.utils.dbsync_utils.check_action_data(*, json_anchor_file: dict[str, Any], anchor_data_hash: str, action_txid: str, voting_anchor_id: int | None = None) → None

Compare anchor json file with off chain action’s data from db-sync.

The anchor is expected to be valid and CIP-conformant: is_valid = TRUE is verified, and authors, references and external updates are populated only for such anchors.

The check targets the gov action’s own off_chain_vote_data row, as the same content can back several anchors. The function waits for the off-chain data row to appear (db-sync’s fetch loop runs every 300s) and is a no-op when db-sync is not available, so callers do not need any db-sync related logic of their own.

Parameters:

• json_anchor_file – Content of the anchor file, to compare the db-sync data against.

• anchor_data_hash – Hash of the anchor content.

• action_txid – Transaction hash of the transaction that submitted the proposal. Used to resolve voting_anchor_id, so it is consulted only when the id is not provided.

• voting_anchor_id – Id of the anchor’s voting_anchor row. Pass it when it is already known, otherwise it is resolved from the proposal’s transaction.

Raises:

• AssertionError – When the data doesn’t match or the row is not stored within the wait timeout.

• DbSyncTimeoutError – When the gov action proposal does not appear in db-sync while resolving voting_anchor_id.

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

Check reward data for stake address in db-sync.

The epoch_from and epoch_to are epochs where the reward can be spent. Return None if db-sync is not available.

cardano_node_tests.utils.dbsync_utils.check_column_condition(*, table: str, column: str, condition: str, expected_count: int | None = None, lock_timeout: str = '5s') → None

Validate that all/none rows meet a column condition atomically.

The check is a no-op when db-sync is not available.

Parameters:

• table – Table to check

• column – Column to validate

• condition – SQL condition (e.g., “= 0”, “IS NOT NULL”)

• expected_count – Require exactly this many matches (default: all must match)

• lock_timeout – Max wait for table lock

cardano_node_tests.utils.dbsync_utils.check_committee_info(*, gov_state: dict, txid: str, action_ix: int = 0) → None

Check committee info in db-sync.

cardano_node_tests.utils.dbsync_utils.check_committee_member_deregistration(*, cc_member_cold_key: str) → CommitteeDeregistrationRecord | None

Check committee member deregistration in db-sync.

cardano_node_tests.utils.dbsync_utils.check_committee_member_registration(*, cc_member_cold_key: str) → CommitteeRegistrationRecord | None

Check committee member registration in db-sync.

cardano_node_tests.utils.dbsync_utils.check_conway_gov_action_proposal_description(*, update_proposal: dict, txhash: str = '', action_ix: int = 0) → GovActionProposalDBRow | None

Check expected values in the gov_action_proposal table in db-sync.

cardano_node_tests.utils.dbsync_utils.check_conway_param_update_enactment(*, pparams: dict, epoch_no: int) → EpochParamDBRow | None

Check params enactment between ledger and epoch param in db-sync.

cardano_node_tests.utils.dbsync_utils.check_conway_param_update_proposal(*, param_proposal_ledger: dict) → ParamProposalDBRow | None

Check comparison for param proposal between ledger and db-sync.

cardano_node_tests.utils.dbsync_utils.check_delegation_vote(*, txhash: str, stake_address: str, drep: str) → None

Check delegation vote in dbsync.

cardano_node_tests.utils.dbsync_utils.check_drep_deregistration(*, drep: DRepRegistration) → DrepRegistrationRecord | None

Check drep deregistration in db-sync.

cardano_node_tests.utils.dbsync_utils.check_drep_registration(*, drep: DRepRegistration, drep_state: list[list[dict[str, Any]]]) → DrepRegistrationRecord | None

Check drep registration in db-sync.

cardano_node_tests.utils.dbsync_utils.check_epoch_state(*, epoch_no: int, txid: str, action_type: ActionTypes) → None

Check governance stats per epoch in dbsync.

cardano_node_tests.utils.dbsync_utils.check_off_chain_drep_registration(*, drep_data: DrepRegistrationRecord, metadata: dict) → None

Check DRep off chain data in db-sync.

cardano_node_tests.utils.dbsync_utils.check_off_chain_vote_fetch_error(*, voting_anchor_id: int) → None

Check expected error in off_chain_vote_fetch_error.

cardano_node_tests.utils.dbsync_utils.check_plutus_cost(*, redeemer_record: RedeemerRecord, cost_record: dict[str, Any]) → None

Compare cost of Plutus script with data from db-sync.

The check is a no-op when db-sync is not available.

cardano_node_tests.utils.dbsync_utils.check_plutus_costs(*, redeemer_records: list[RedeemerRecord], cost_records: list[dict[str, Any]]) → None

Compare cost of multiple Plutus scripts with data from db-sync.

The check is a no-op when db-sync is not available.

cardano_node_tests.utils.dbsync_utils.check_pool_data(*, ledger_pool_data: dict, pool_id: str) → PoolDataRecord | None

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) → PoolDataRecord | None

Check pool retirement in db-sync.

cardano_node_tests.utils.dbsync_utils.check_pool_off_chain_data(*, ledger_pool_data: dict, pool_id: str) → PoolOffChainDataDBRow | None

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

Return None if db-sync is not available.

cardano_node_tests.utils.dbsync_utils.check_pool_off_chain_fetch_error(*, ledger_pool_data: dict, pool_id: str) → PoolOffChainFetchErrorDBRow | None

Check expected error on OffChainPoolFetchError.

Return None if db-sync is not available.

cardano_node_tests.utils.dbsync_utils.check_proposal_refunds(*, stake_address: str, refunds_num: int) → None

Check proposal refunds in db-sync.

cardano_node_tests.utils.dbsync_utils.check_reward_rest(*, stake_address: str, transfer_amts: list[int], type: str = '') → None

Check reward_rest in db-sync.

cardano_node_tests.utils.dbsync_utils.check_treasury_withdrawal(*, stake_address: str, transfer_amts: list[int], txhash: str) → None

Check treasury_withdrawal in db-sync.

cardano_node_tests.utils.dbsync_utils.check_tx(*, cluster_obj: ClusterLib, tx_raw_output: TxRawOutput, retry_num: int = 3) → TxRecord | None

Check a transaction in db-sync.

cardano_node_tests.utils.dbsync_utils.check_tx_phase_2_failure(*, cluster_obj: ClusterLib, tx_raw_output: TxRawOutput, collateral_charged: int, retry_num: int = 3) → TxRecord | None

Check a transaction in db-sync when a phase 2 failure happens.

cardano_node_tests.utils.dbsync_utils.check_votes(*, votes: VotedVotes, txhash: str) → None

Check votes in db-sync.

cardano_node_tests.utils.dbsync_utils.get_action_data(*, data_hash: str, voting_anchor_id: int) → OffChainVoteDataRecord | None

Get off chain action data from db-sync.

Parameters:

• data_hash – Hash of the anchor content.

• voting_anchor_id – Id of the anchor’s voting_anchor row. The same content (same data_hash) can back several anchors of different types, each with its own row, so the target anchor must be identified explicitly.

Returns:

The off-chain vote data record, or None when db-sync has no row for the hash / voting_anchor_id combination (e.g. the anchor was not fetched yet).

cardano_node_tests.utils.dbsync_utils.get_address_reward(address: str, *, epoch_from: int = 0, epoch_to: int = 99999999) → 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) → CommitteeRegistrationRecord | None

Get committee member data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_deregistered_committee_member(*, cold_key: str) → CommitteeDeregistrationRecord | None

Get deregistered committee member data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_drep(*, drep_hash: str, drep_deposit: int) → DrepRegistrationRecord | None

Get drep data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_gov_action_proposals(*, txhash: str = '', type: str = '') → list[GovActionProposalDBRow]

Get government action proposal from db-sync.

cardano_node_tests.utils.dbsync_utils.get_gov_action_voting_anchor_id(*, txhash: str, timeout: int = 120) → int

Return the voting_anchor_id db-sync assigned to a gov action proposal.

The same anchor content can back several anchors (e.g. a gov action and a DRep registration), so callers that assert anchor details use this to target the gov action’s own off_chain_vote_data row. The transaction is expected to carry a single proposal; if it carries more, an arbitrary one is used.

Parameters:

• txhash – Transaction hash of the transaction that submitted the proposal.

• timeout – Lower bound on how long to wait for the proposal to appear in db-sync, in seconds (retry_query backoff can overshoot it).

Raises:

DbSyncTimeoutError – When the proposal does not appear in db-sync within the timeout.

cardano_node_tests.utils.dbsync_utils.get_pool_data(*, pool_id_bech32: str) → PoolDataRecord | None

Get pool data from db-sync.

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

Get first batch of transaction data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_tx(*, cluster_obj: ClusterLib, tx_raw_output: TxRawOutput, retry_num: int = 3) → TxRecord | None

Get a transaction data from db-sync.

cardano_node_tests.utils.dbsync_utils.get_tx_record(*, txhash: str) → 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) → 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[UTxORecord]

Get txins of a transaction from db-sync.

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

Return UTxO info for payment address from db-sync.

cardano_node_tests.utils.dbsync_utils.map_params_to_db_convention(*, pparams: dict) → dict[str, Any]

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 DbSyncNoResponseError. The query is repeated until the expected data is returned, or DbSyncTimeoutError is raised when the timeout is reached.

cardano_node_tests.utils.dbsync_utils.table_empty(*, table: str) → bool

Check if a database table is empty.

cardano_node_tests.utils.dbsync_utils.table_exists(*, table: str) → bool

Check if a table exists in the database.

cardano_node_tests.utils.dbsync_utils.wait_for_db_sync_completion(*, expected_progress: float = 99.0, timeout: int = 360, polling_interval: int = 5) → float

Wait for db-sync to reach the expected sync completion.

Parameters:

• expected_progress – Expected completion as percentage, 99% by default

• timeout – Maximum total time to wait in seconds

• polling_interval – Loop polling time in seconds

Returns:

Final sync percentage achieved (>= expected_progress)

Raises:

DbSyncTimeoutError – If sync doesn’t reach the expected progress within timeout

cardano_node_tests.utils.dbsync_utils.wait_for_off_chain_vote_data(*, data_hash: str, voting_anchor_id: int, timeout: int = 420) → OffChainVoteDataRecord

Wait until db-sync stores the off-chain vote data for an anchor and return it.

db-sync’s off-chain fetch loop sleeps 300s between passes, so the row can appear up to ~5 minutes after the anchor is on-chain.

A missing row can mean a db-sync regression (the PR #2005 behaviour), a failed download (404, network trouble - recorded by db-sync in off_chain_vote_fetch_error), or a wrong voting_anchor_id. On timeout, the failure message includes the recorded fetch errors and a data-presence diagnostic (whether the data appeared late, which other anchor ids hold data for the hash, or that none do), to tell the cases apart.

Parameters:

• data_hash – Hash of the anchor content.

• voting_anchor_id – Id of the anchor’s voting_anchor row. The same content can be referenced by several anchors (e.g. a gov action and a DRep registration share a data_hash), so the target anchor must be identified explicitly.

• timeout – Lower bound on how long to wait for the row, in seconds. retry_query uses quadratic backoff and checks the deadline only between polls, so the actual wait can exceed this by up to one backoff interval (~25% at the default 420s).

Returns:

The stored off-chain vote data record.

Raises:

AssertionError – When the row is not stored within the timeout.

cardano_node_tests.utils.defragment_utxos module

Defragment address UTxOs.

cardano_node_tests.utils.defragment_utxos.defragment(*, cluster_obj: ClusterLib, address: str, skey_file: Path, max_len: int = 10, name_template: str = '') → None

Defragment address UTxOs.

cardano_node_tests.utils.dev_session module

Exclusive use of a dev cluster by a single pytest invocation.

With DEV_CLUSTER_RUNNING, multiple concurrent pytest invocations (possibly from different git worktrees) can target the same dev cluster instance. The cluster management locks live under the invocation-specific pytest temp dir, so they cannot coordinate across invocations. Instead, each invocation records its identity in a claim file next to the dev cluster state dir, and refuses to run when the cluster is already claimed by another live invocation.

cardano_node_tests.utils.dev_session.claim_session() → None

Claim exclusive use of the dev cluster for this pytest invocation.

Raises:

RuntimeError – When the dev cluster is already used by another live pytest invocation, or when the session lock cannot be acquired.

cardano_node_tests.utils.dev_session.release_session() → None

Release the dev cluster claimed by this pytest invocation.

Best-effort - failures are logged but never raised, so a failed release cannot mask cleanup errors in the session teardown.

cardano_node_tests.utils.faucet module

cardano_node_tests.utils.faucet.fund_from_faucet(*dst_addrs: AddressRecord, cluster_obj: ClusterLib, faucet_data: dict | None = None, all_faucets: dict[str, dict] | None = None, amount: int | list[int], tx_name: str | None = None, destination_dir: str | Path = '.', force: bool = False) → TxRawOutput | None

Transfer amount from faucet addr to all dst_addrs.

cardano_node_tests.utils.faucet.return_funds_to_faucet(*src_addrs: AddressRecord, cluster_obj: ClusterLib, faucet_addr: str, amount: int | list[int] = -1, tx_name: str | None = None, destination_dir: str | Path = '.') → None

Transfer amount from all src_addrs to faucet_addr.

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

cardano_node_tests.utils.framework_log module

cardano_node_tests.utils.framework_log.framework_logger() → 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.framework_log.get_framework_log_path() → Path

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[str | None] = 'token_XXXXXXXXXXXXXXXXXXXX'

get_state() → str

Get issue state.

Returns:

The state as reported by GitHub, e.g. “open” or STATE_CLOSED; STATE_UNKNOWN when the issue cannot be found, or STATE_FAILURE when the state could not be retrieved.

property github: Github | None

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

property url: str

cardano_node_tests.utils.governance_setup module

cardano_node_tests.utils.governance_setup.auth_cc_members(*, cluster_obj: ClusterLib, cc_members: list[CCKeyMember], name_template: str, payment_addr: AddressRecord) → None

Authorize the original CC members.

cardano_node_tests.utils.governance_setup.create_vote_stake(*, name_template: str, cluster_manager: ClusterManager, cluster_obj: ClusterLib, no_of_addr: int, destination_dir: str | Path = '.') → list[PoolUser]

cardano_node_tests.utils.governance_setup.get_default_governance(*, cluster_manager: ClusterManager, cluster_obj: ClusterLib) → GovernanceRecords

Get default governance data for CC members, DReps and SPOs.

cardano_node_tests.utils.governance_setup.load_committee(*, cluster_obj: ClusterLib) → list[CCKeyMember]

cardano_node_tests.utils.governance_setup.load_drep_users(*, cluster_obj: ClusterLib) → list[PoolUser]

Load DReps users from the state directory.

cardano_node_tests.utils.governance_setup.load_dreps(*, cluster_obj: ClusterLib) → list[DRepRegistration]

Load DReps from the state directory.

cardano_node_tests.utils.governance_setup.refresh_cc_keys(*, cluster_obj: ClusterLib, cc_members: list[CCKeyMember], governance_data: GovernanceRecords) → GovernanceRecords

Refresh hot certs for original CC members.

cardano_node_tests.utils.governance_setup.reinstate_committee(*, cluster_obj: ClusterLib, governance_data: GovernanceRecords, name_template: str, pool_user: PoolUser) → None

Reinstate the original CC members.

cardano_node_tests.utils.governance_setup.save_default_governance(*, dreps_reg: list[DRepRegistration], drep_delegators: list[PoolUser], cc_members: list[CCKeyMember], pools_cold: list[ColdKeyPair]) → GovernanceRecords

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: ClusterManager, cluster_obj: ClusterLib) → GovernanceRecords

cardano_node_tests.utils.governance_utils module

Utilities for Conway governance.

class cardano_node_tests.utils.governance_utils.ActionTags(*values)

Bases: Enum

HARDFORK_INIT = 'HardForkInitiation'

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.AnchorData(url: str, hash: str, data_file: pathlib.Path | None)

Bases: object

data_file: Path | None

hash: str

url: str

class cardano_node_tests.utils.governance_utils.CCHotKeys(hot_vkey: str = '', hot_vkey_file: str | pathlib.Path = '', hot_vkey_hash: str = '', hot_skey: str = '', hot_skey_file: str | pathlib.Path = '', hot_skey_hash: str = '')

Bases: object

hot_skey: str = ''

hot_skey_file: str | Path = ''

hot_skey_hash: str = ''

hot_vkey: str = ''

hot_vkey_file: str | Path = ''

hot_vkey_hash: str = ''

class cardano_node_tests.utils.governance_utils.CCKeyMember(cc_member: cardano_clusterlib.structs.CCMember, hot_keys: cardano_node_tests.utils.governance_utils.CCHotKeys)

Bases: object

cc_member: CCMember

hot_keys: CCHotKeys

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: Path

cold_key_pair: KeyPair

hot_key_pair: 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: KeyPair

registration_cert: Path

class cardano_node_tests.utils.governance_utils.DRepScriptRegInputs(registration_cert: cardano_clusterlib.structs.ComplexCert, key_pairs: list[cardano_clusterlib.structs.KeyPair], script_type: cardano_node_tests.utils.governance_utils.ScriptTypes)

Bases: object

key_pairs: list[KeyPair]

registration_cert: ComplexCert

script_type: ScriptTypes

class cardano_node_tests.utils.governance_utils.DRepScriptRegRecord(registration_cert: pathlib.Path, script_hash: str, deposit: int)

Bases: object

deposit: int

registration_cert: Path

script_hash: str

class cardano_node_tests.utils.governance_utils.DRepScriptRegistration(registration_cert: cardano_clusterlib.structs.ComplexCert, key_pairs: list[cardano_clusterlib.structs.KeyPair], script_hash: str, script_type: cardano_node_tests.utils.governance_utils.ScriptTypes, deposit: int)

Bases: object

deposit: int

key_pairs: list[KeyPair]

registration_cert: ComplexCert

script_hash: str

script_type: ScriptTypes

class cardano_node_tests.utils.governance_utils.GovernanceRecords(dreps_reg: list[cardano_node_tests.utils.governance_utils.DRepRegistration], drep_delegators: list[cardano_clusterlib.structs.PoolUser], cc_key_members: list[cardano_node_tests.utils.governance_utils.CCKeyMember], pools_cold: list[cardano_clusterlib.structs.ColdKeyPair], drep_scripts_reg: list[cardano_node_tests.utils.governance_utils.DRepScriptRegistration] = <factory>, drep_scripts_delegators: list[cardano_clusterlib.structs.PoolUser] = <factory>)

Bases: object

cc_key_members: list[CCKeyMember]

drep_delegators: list[PoolUser]

drep_scripts_delegators: list[PoolUser]

drep_scripts_reg: list[DRepScriptRegistration]

dreps_reg: list[DRepRegistration]

pools_cold: list[ColdKeyPair]

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(*values)

Bases: Enum

COMMITTEE = 'Committee'

CONSTITUTION = 'Constitution'

HARDFORK = 'HardFork'

PPARAM_UPDATE = 'PParamUpdate'

class cardano_node_tests.utils.governance_utils.ScriptTypes(*values)

Bases: Enum

PLUTUS = 'plutus'

SIMPLE = 'simple'

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

Bases: object

drep: int

spo: int

total_lovelace: int

class cardano_node_tests.utils.governance_utils.VotedVotes(cc: list[cardano_clusterlib.structs.VoteCC], drep: list[cardano_clusterlib.structs.VoteDrep], spo: list[cardano_clusterlib.structs.VoteSPO])

Bases: object

cc: list[VoteCC]

drep: list[VoteDrep]

spo: list[VoteSPO]

cardano_node_tests.utils.governance_utils.check_action_view(*, cluster_obj: ClusterLib, action_data: ActionConstitution | ActionHardfork | ActionInfo | ActionNoConfidence | ActionPParamsUpdate | ActionTreasuryWithdrawal | ActionUpdateCommittee, 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: dict[str, Any], drep_id: str, min_amount: int) → None

cardano_node_tests.utils.governance_utils.check_vote_view(*, cluster_obj: ClusterLib, vote_data: VoteCC | VoteDrep | VoteSPO) → None

Check governance vote view output.

cardano_node_tests.utils.governance_utils.create_dreps(*, name_template: str, num: int, cluster_obj: ClusterLib, payment_addr: AddressRecord, pool_users: list[PoolUser], destination_dir: str | Path = '.') → tuple[list[DRepRegistration], list[PoolUser]]

Create DReps with keys.

cardano_node_tests.utils.governance_utils.create_script_dreps(*, name_template: str, script_inputs: list[DRepScriptRegInputs], cluster_obj: ClusterLib, payment_addr: AddressRecord, pool_users: list[PoolUser], destination_dir: str | Path = '.') → tuple[list[DRepScriptRegistration], list[PoolUser]]

Create DReps with scripts.

cardano_node_tests.utils.governance_utils.get_anchor_data(*, cluster_obj: ClusterLib, name_template: str, anchor_text: str) → AnchorData

Publish anchor data and return the URL and data hash.

cardano_node_tests.utils.governance_utils.get_cc_member_auth_record(*, cluster_obj: ClusterLib, name_template: str, destination_dir: str | Path = '.') → CCMemberAuth

Get Constitutional Committee Members key authorization record.

cardano_node_tests.utils.governance_utils.get_default_anchor_data() → AnchorData

Return the default anchor data.

cardano_node_tests.utils.governance_utils.get_delegated_stake(*, cluster_obj: ClusterLib) → 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_cred_name_from_addr_info(*, addr_info: StakeAddrInfo) → str

cardano_node_tests.utils.governance_utils.get_drep_reg_record(*, cluster_obj: ClusterLib, name_template: str, deposit_amt: int = -1, drep_metadata_url: str = '', drep_metadata_hash: str = '', destination_dir: str | Path = '.') → DRepRegistration

Get DRep registration record.

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

cardano_node_tests.utils.governance_utils.get_script_drep_reg_record(*, cluster_obj: ClusterLib, name_template: str, script_hash: str, deposit_amt: int = -1, drep_metadata_url: str = '', drep_metadata_hash: str = '', destination_dir: str | Path = '.') → DRepScriptRegRecord

Get DRep script registration record.

cardano_node_tests.utils.governance_utils.get_vote_str(*, vote: Votes) → str

cardano_node_tests.utils.governance_utils.is_cc_active(*, cc_member_state: dict[str, Any]) → bool

Check if CC member is active.

cardano_node_tests.utils.governance_utils.is_drep_active(*, cluster_obj: ClusterLib, drep_state: list[list[dict[str, Any]]], epoch: int = -1) → bool

Check if DRep is active.

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(*, 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: ClusterLib) → None

Wait until ratification is no longer delayed.

cardano_node_tests.utils.helpers module

exception cardano_node_tests.utils.helpers.CommandExecError

Bases: RuntimeError

Raised when a command cannot be executed at all (e.g. missing executable or workdir).

cardano_node_tests.utils.helpers.change_cwd(dir_path: str | Path) → Generator[str | Path, None, None]

Change and restore CWD - context manager.

cardano_node_tests.utils.helpers.check_cardano_node_socket_path() → None

Check that CARDANO_NODE_SOCKET_PATH value is valid for use by testing framework.

cardano_node_tests.utils.helpers.check_dir_arg(dir_path: str) → Path | None

Check that the dir passed as argparse parameter is a valid existing dir.

cardano_node_tests.utils.helpers.check_dir_arg_keep(dir_path: str) → Path | None

Check that the dir passed as argparse parameter is a valid existing dir.

Keep the original path (with ~ expanded) instead of resolving it to absolute.

cardano_node_tests.utils.helpers.check_file_arg(file_path: str) → Path | None

Check that the file passed as argparse parameter is a valid existing file.

cardano_node_tests.utils.helpers.checksum(filename: str | Path) → 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[str, str]) → Generator[None, None, None]

Temporarily set environment variables and restore previous environment afterwards.

cardano_node_tests.utils.helpers.flatten(iterable: Iterable[Any], *, ltypes: type[Iterable[Any]] | None = None) → Generator[Any, None, None]

Flatten an irregular (arbitrarily nested) iterable of iterables.

cardano_node_tests.utils.helpers.get_current_commit() → str

Return the current git commit hash.

Uses the GIT_REVISION env variable when set to a non-empty value, otherwise asks git. Git is run in the repo containing this module, independent of CWD. The result is cached for the lifetime of the process.

cardano_node_tests.utils.helpers.get_current_line_str() → str

Get filename#L<lineno> of current line.

NOTE: Reports the location of the immediate caller. Calling through any wrapper (decorator, context manager, functools.partial) reports the wrapper’s location instead.

cardano_node_tests.utils.helpers.get_env_int(var: str, default: int) → int

Read an integer environment variable.

Returns default when the variable is unset or empty. Raises a ValueError naming the offending variable when the value is malformed.

cardano_node_tests.utils.helpers.get_env_path(var: str) → Path | None

Read a path environment variable.

Returns the resolved pathlib.Path (with ~ expanded) when set, otherwise None.

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

Return filename#L<lineno> string for the given frame.

cardano_node_tests.utils.helpers.get_pool_param(key: str, *, pool_params: dict) → Any

Get pool parameter value from pool params dict.

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() → Generator[None, None, 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.is_inputs_spent_err(err_str: str) → bool

Check if the error message indicates that the Tx inputs were already spent.

The ledger reports this condition with a different error in each era, so all the known variants are accepted. In the Dijkstra era the Conway mempool failure is injected as the dedicated AllInputsAreSpent failure.

Parameters:

err_str – An error message to check.

Returns:

True if the error was caused by already spent Tx inputs.

Return type:

bool

cardano_node_tests.utils.helpers.is_truthy_env_var(var: str) → bool

Check if an environment variable is set to a truthy value.

cardano_node_tests.utils.helpers.prepend_flag(flag: str, contents: Iterable[Any]) → 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: str | list[Any], *, workdir: str | Path = '', ignore_fail: bool = False, shell: bool = False, merge_stderr: bool = False, stdin_data: bytes | None = None) → bytes

Run command.

Parameters:

• command – A command to run - either a string (tokenized with shlex unless shell is used) or a list of arguments (items are converted to str).

• workdir – A working directory for the command.

• ignore_fail – Don’t raise an exception when the command fails.

• shell – Run the command in a shell.

• merge_stderr – Redirect stderr to stdout.

• stdin_data – Data to pass to stdin of the command.

Returns:

Content of stdout (with stderr merged in when merge_stderr is used).

Return type:

bytes

Raises:

• RuntimeError – When the command fails (unless ignore_fail is used).

• CommandExecError – When the command cannot be executed at all (e.g. missing or non-executable file). Subclass of RuntimeError.

cardano_node_tests.utils.helpers.run_in_bash(command: str, *, workdir: str | 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(“cardano-cli legacy governance”)

Raises:

CommandExecError – When the tool cannot be executed at all, as its availability cannot be determined in that case.

cardano_node_tests.utils.helpers.validate_dict_values(dict1: dict[str, Any], dict2: dict[str, Any], *, keys: Iterable[str]) → list[str]

Compare values for specified keys between two dictionaries and return discrepancies.

Parameters:

• dict1 – First dictionary to compare. This represents the expected data.

• dict2 – Second dictionary to compare. This represents the actual data.

• keys – List of keys to compare between the two dictionaries.

Returns:

A list of discrepancies, with each discrepancy describing a mismatch between the values in dict1 and dict2 for the specified keys.

cardano_node_tests.utils.helpers.write_json(*, out_file: str | Path, content: dict) → str | Path

Write dictionary content to JSON file.

cardano_node_tests.utils.http_client module

Global HTTP client.

cardano_node_tests.utils.http_client.get_session() → Session

Get a session object.

cardano_node_tests.utils.locking module

File-based locking.

FileLock is a real lock that works regardless of how pytest is invoked - use it when coordination is needed even outside pytest-xdist, e.g. across pytest invocations. FileLockIfXdist degrades to a no-op when not running with pytest-xdist workers.

cardano_node_tests.utils.locking.FileLock

alias of UnixFileLock

cardano_node_tests.utils.locking.FileLockIfXdist

alias of nullcontext

exception cardano_node_tests.utils.locking.Timeout(lock_file: str)

Bases: TimeoutError

Raised when the lock could not be acquired in timeout seconds.

property lock_file: str

The path of the file lock.

Type:

return

cardano_node_tests.utils.logfiles module

class cardano_node_tests.utils.logfiles.RotableLog(logfile: pathlib.Path, seek: int, timestamp: float, inode: int)

Bases: object

inode: int

logfile: 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.

Raises:

ValueError – When the rule cannot be stored in the rules file format (“;;” in the file glob, a line break in the file glob or in the regex), when the regex is not valid or cannot be combined with other rules, when the regex matches an empty string (it would suppress all errors for the matching files), or when the skip_after value is not a non-negative finite number.

NOTE: Conflicts between rules that fail compilation of the combined ignore regex (e.g. two rules redefining the same group name) cannot be detected here. Such rules are skipped with a warning when the combined ignore regex is built. Avoid numeric backreferences (e.g. “1”) - group numbers shift in the combined regex, so the backreference can silently bind to a group of a different rule.

cardano_node_tests.utils.logfiles.check_msgs_presence_in_logs(regex_pairs: list[tuple[str, str]], seek_offsets: Mapping[str, tuple[int, int | None]], state_dir: Path, timestamp: float) → list[str]

Check if the expected messages are present in logs (byte offsets, binary I/O).

An expected message found in an unterminated final line counts as present, unless the regex constrains what follows the match (see _constrains_match_end).

Parameters:

• regex_pairs – (glob, regex) pairs.

• seek_offsets – Mapping of absolute log path (str) -> (byte offset, inode or None).

• state_dir – Root directory for globs.

• timestamp – Passed to _get_rotated_logs.

Returns:

Error messages for missing entries, for globs that matched no log file, and for log files that could not be searched.

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) → Generator[None, None, None]

Make sure the expected errors are present in logs.

Context manager.

Parameters:

• regex_pairs – [(glob, regex)] - A list of regexes matching strings 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.expect_messages(regex_pairs: list[tuple[str, str]]) → Generator[None, None, None]

Make sure the expected messages are present in logs.

Context manager.

Parameters:

regex_pairs – [(glob, regex)] - A list of regexes matching strings that need to be present in files described by the glob.

cardano_node_tests.utils.logfiles.find_msgs_in_logs(*, regex: str, logfile: Path, seek_offset: int, timestamp: float, inode: int | None = None, only_first: bool = False, encoding: str = 'utf-8') → list[str]

Find messages in log using a byte seek offset (binary mode).

Reads in binary for speed, resumes at a line boundary near seek_offset, matches with a bytes regex, and decodes only matching lines.

Parameters:

• regex – Regular expression (interpreted as UTF-8 when compiled to bytes).

• logfile – Path to the primary log file.

• seek_offset – Byte offset to resume from (start of a line or 0).

• timestamp – Timestamp used by _get_rotated_logs(…).

• inode – Inode of the log file the seek_offset was recorded for (optional).

• only_first – If True, return immediately after the first match.

• encoding – Text encoding used to decode matched lines (default: “utf-8”).

Returns:

Matching log lines (decoded to str, without trailing newlines). An unterminated final line of the live log file is not searched, so a truncated line is never returned for it. An unterminated final line of a rotated log file is searched, as it is the file’s final content.

Raises:

FileNotFoundError – When the log file keeps getting rotated during the search. Unlike the error-aggregating searches, the failure is propagated - the caller needs the matching lines, and an empty result would be misread as “no matches”.

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[Path, str]]

Search cluster logs for errors.

cardano_node_tests.utils.logfiles.search_framework_log() → list[tuple[Path, str]]

Search framework log for errors.

cardano_node_tests.utils.logfiles.search_supervisord_logs() → list[tuple[Path, str]]

Search cluster logs for errors.

cardano_node_tests.utils.model_ekg module

class cardano_node_tests.utils.model_ekg.Adopted(*, int: Int7)

Bases: BaseModel

int: Int7

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int1

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int2

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

node: Node

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

Bases: BaseModel

int: Int20

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

int: Int29

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

int: Int9

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int18

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

real: Real

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

server_timestamp_ms: ServerTimestampMs

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

Bases: BaseModel

int: Int3

model_config: ClassVar[ConfigDict] = {}

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

class cardano_node_tests.utils.model_ekg.Forge(*, node_is_leader: NodeIsLeader, node_not_leader: NodeNotLeader, forged: Forged, adopted: Adopted, forge_about_to_lead: ForgeAboutToLead)

Bases: 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].

node_is_leader: NodeIsLeader

node_not_leader: NodeNotLeader

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

Bases: BaseModel

int: Int8

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int6

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: 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].

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: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

int: Int23

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int22

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int24

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

int: Int26

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

counter: Counter

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: str

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

resident: Resident

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

Bases: BaseModel

int: Int21

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: 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].

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(*, iohk_monitoring_version: IohkMonitoringVersion, cardano: Cardano, ekg: Ekg, rts: Rts)

Bases: 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].

rts: Rts

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

int: Int25

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int10

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

metrics: Metrics

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int4

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int5

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int12

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

int: Int15

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int14

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: 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].

mutticks: Mutticks

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: str

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

Bases: BaseModel

int: Int16

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int13

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

gc: Gc

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

header: Header

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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

Bases: BaseModel

int: Int27

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int28

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

cputicks: Cputicks

model_config: ClassVar[ConfigDict] = {}

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

threads: Threads

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

Bases: BaseModel

int: Int30

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int17

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int11

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

int: Int19

model_config: ClassVar[ConfigDict] = {}

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

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

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

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

type: str

val: int

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: Path

test_function: str

test_params: str = ''

cardano_node_tests.utils.pytest_utils.get_current_test() → 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(*values)

Bases: StrEnum

CHANG_US = 'chang_us'

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

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

success() → bool

class cardano_node_tests.utils.requirements.Statuses(*values)

Bases: Enum

failure = 2

partial_success = 3

success = 1

uncovered = 4

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

Collect executed requirements.

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

Get mapped requirements.

cardano_node_tests.utils.requirements.merge_reqs(*reqs: dict[str, dict]) → dict

Merge requirements.

cardano_node_tests.utils.smash_utils module

class cardano_node_tests.utils.smash_utils.PoolData(pool_id: str)

Bases: object

pool_id: str

class cardano_node_tests.utils.smash_utils.PoolError(cause: str, pool_hash: str, pool_id: str, retry_count: int, time: str, utc_time: str)

Bases: object

cause: str

pool_hash: str

pool_id: str

retry_count: int

time: str

utc_time: str

class cardano_node_tests.utils.smash_utils.PoolMetadata(name: str, description: str, ticker: str, homepage: str)

Bases: object

description: str

homepage: str

name: str

ticker: str

class cardano_node_tests.utils.smash_utils.PoolTicker(name: str)

Bases: object

name: str

class cardano_node_tests.utils.smash_utils.SmashClient(instance_num: int, req_timeout: int = 10)

Bases: object

Utility class for interacting with SMASH via REST API.

delist_pool(*, pool_id: str) → PoolData

Delist a stake pool, returning PoolData on success or a RequestException on failure.

enlist_pool(*, pool_id: str) → PoolData

Enlist a stake pool, returning PoolData on success or a RequestException on failure.

get_pool_errors(*, pool_id: str, from_date: str | None = None) → list[PoolError]

Fetch errors for a specific stake pool.

get_pool_metadata(*, pool_id: str, pool_meta_hash: str) → PoolMetadata

Fetch stake pool metadata from SMASH, returning a PoolMetadata.

get_retired_pools() → list[PoolData]

Fetch list of retired pools.

reserve_ticker(*, ticker_name: str, pool_hash: str) → PoolTicker

Reserve a ticker for a stake pool.

class cardano_node_tests.utils.smash_utils.SmashManager

Bases: object

classmethod get_smash_instance() → SmashClient

Get a singleton instance of SmashClient for the given cluster instance.

instances: ClassVar[dict[int, SmashClient]] = {}

cardano_node_tests.utils.smash_utils.check_smash_pool_errors(*, pool_id: str, pool_metadata_hash: str) → list[PoolError] | None

Check if pool errors are correctly reported in SMASH.

cardano_node_tests.utils.smash_utils.check_smash_pool_retired(*, pool_id: str) → list[PoolData] | None

Check if pool is correctly reported as retired in SMASH.

cardano_node_tests.utils.smash_utils.get_client() → SmashClient | None

Get and cache the SMASH client.

cardano_node_tests.utils.smash_utils.is_smash_running() → bool

Check if cardano-smash-server service is running.

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: 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: str | Path, url: str) → Response

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

cardano_node_tests.utils.submit_api.submit_tx(*, cluster_obj: ClusterLib, tx_file: str | Path, txins: list[UTXOData], wait_blocks: int = 2) → str

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: str | Path) → SubmitApiOut

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

cardano_node_tests.utils.submit_api.tx2cbor(*, tx_file: str | Path, destination_dir: str | Path = '.') → Path

Convert signed Tx to binary CBOR.

cardano_node_tests.utils.submit_utils module

class cardano_node_tests.utils.submit_utils.SubmitMethods(*values)

Bases: StrEnum

API = 'api'

CLI = '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: ClusterLib, tx_file: str | Path, txins: list[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: TempPathFactory) → None

pytest_root_tmp: ClassVar[Path | None] = None

pytest_shared_tmp: ClassVar[Path | None] = None

pytest_worker_tmp: ClassVar[Path | None] = None

cardano_node_tests.utils.temptools.get_basetemp() → Path

Return base temporary directory for tests artifacts.

cardano_node_tests.utils.temptools.get_pytest_root_tmp() → Path

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

cardano_node_tests.utils.temptools.get_pytest_shared_tmp() → 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() → 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

• Retire DReps

• Return funds to faucet

cardano_node_tests.utils.testnet_cleanup.addresses_info(*, cluster_obj: ClusterLib, location: Path) → Tuple[int, int]

Return the total balance and rewards of all addresses in the given location.

cardano_node_tests.utils.testnet_cleanup.batch_tx_inputs(tx_inputs: list[tuple[list[UTXOData], Path]], *, batch_size: int = 100) → Generator[list[tuple[list[UTXOData], Path]], None, None]

Batch transaction inputs.

cardano_node_tests.utils.testnet_cleanup.cleanup(*, cluster_obj: ClusterLib, location: str | Path, faucet_address: str = '', faucet_skey_file: str | Path = '') → None

Cleanup a testnet with the help of testing artifacts.

cardano_node_tests.utils.testnet_cleanup.cleanup_addresses(*, cluster_obj: ClusterLib, location: Path, faucet_payment: AddressRecord) → None

Cleanup addresses.

cardano_node_tests.utils.testnet_cleanup.cleanup_certs(*, cluster_obj: ClusterLib, location: Path, faucet_payment: AddressRecord) → None

Cleanup certificates.

cardano_node_tests.utils.testnet_cleanup.create_addr_record(*, addr_file: Path) → AddressRecord

Return a clusterlib.AddressRecord.

cardano_node_tests.utils.testnet_cleanup.dedup_addresses(file_paths: Iterator[Path]) → Iterator[Path]

Deduplicate addresses.

cardano_node_tests.utils.testnet_cleanup.dedup_tx_inputs(tx_inputs: list[tuple[list[UTXOData], Path]]) → list[tuple[list[UTXOData], Path]]

Deduplicate transaction inputs.

cardano_node_tests.utils.testnet_cleanup.deregister_stake_addr(*, cluster_obj: ClusterLib, payment_addr: AddressRecord, stake_addr: AddressRecord, stake_addr_info: StakeAddrInfo, name_template: str, deposit_amt: int) → None

Deregister stake address.

cardano_node_tests.utils.testnet_cleanup.fees_info(*, cluster_obj: ClusterLib, location: Path) → int

Return the total fees of all signed transactions in the given location.

cardano_node_tests.utils.testnet_cleanup.filter_addr_files(file_paths: Iterator[Path]) → Iterator[Path]

Skip the ‘*_pycurrent’ symlinks to pytest temp dirs and faucet address.

cardano_node_tests.utils.testnet_cleanup.find_addr_files(location: Path) → Iterator[Path]

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

cardano_node_tests.utils.testnet_cleanup.find_cert_files(location: Path) → Iterator[Path]

Find all ‘*_drep_reg.cert’ files in given location and its subdirectories.

cardano_node_tests.utils.testnet_cleanup.find_submitted_tx_files(location: Path) → Iterator[Path]

Find all ‘*.submitted’ files in given location and its subdirectories.

cardano_node_tests.utils.testnet_cleanup.flatten_tx_inputs(tx_inputs: list[tuple[list[UTXOData], Path]]) → Tuple[list[UTXOData], list[Path]]

Flatten transaction inputs.

cardano_node_tests.utils.testnet_cleanup.get_tx_inputs(*, cluster_obj: ClusterLib, src_addrs: list[AddressRecord]) → list[tuple[list[UTXOData], Path]]

Return signing keys and transaction inputs for given addresses.

Exclude UTxOs that contain tokens. Don’t exclude UTxOs with datum. All the UTxOs here has ‘skey’, and so the UTxOs with datum are spendable reference inputs.

cardano_node_tests.utils.testnet_cleanup.is_framework_testnet() → bool

Determine if the testnet is a testnet started by the testing framework.

cardano_node_tests.utils.testnet_cleanup.reregister_stake_addr(*, cluster_obj: ClusterLib, payment_addr: AddressRecord, stake_addr: AddressRecord, stake_addr_info: StakeAddrInfo, name_template: str, deposit_amt: int) → StakeAddrInfo

Re-register stake address.

The address needs to be registered and delegated to a DRep to be able to withdraw rewards.

cardano_node_tests.utils.testnet_cleanup.retire_drep(*, cluster_obj: ClusterLib, payment_addr: AddressRecord, drep_keys: KeyPair, name_template: str, deposit_amt: int) → None

Retire a DRep.

cardano_node_tests.utils.testnet_cleanup.return_funds_to_faucet(*, cluster_obj: ClusterLib, tx_inputs: list[tuple[list[UTXOData], Path]], faucet_address: str, tx_name: str = '') → None

Send funds from tx_inputs to faucet_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: ClusterLib, tx_raw_output: 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: ClusterLib, tx_body_file: 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.EraName(*values)

Bases: StrEnum

Cardano era names.

Members compare equal to their lowercase era name strings (e.g. EraName.CONWAY == "conway"), so they can be used anywhere an era name string is expected.

ALLEGRA = 'allegra'

ALONZO = 'alonzo'

BABBAGE = 'babbage'

BYRON = 'byron'

CONWAY = 'conway'

DIJKSTRA = 'dijkstra'

MARY = 'mary'

SHELLEY = 'shelley'

class cardano_node_tests.utils.versions.Versions

Bases: object

Protocol version, cluster era, transaction era, node version info.

Note

cluster_era holds the current protocol version (set via the PROTOCOL_VERSION env var, defaulting to DEFAULT_CLUSTER_ERA), while cluster_era_name is the era name that corresponds to that protocol version. The era constants below (e.g. CONWAY) are the latest protocol version associated with each era, and the *_FIRST constants (e.g. CONWAY_FIRST) are the first protocol version of each era. Use the *_FIRST constants for era membership checks (e.g. transaction_era >= VERSIONS.CONWAY_FIRST for “in Conway era or later”), so the checks keep working for all protocol versions of an era. Both sets of constants must stay consistent with MAP.

ALLEGRA: Final[int] = 3

ALLEGRA_FIRST: Final[int] = 3

ALONZO: Final[int] = 6

ALONZO_FIRST: Final[int] = 5

BABBAGE: Final[int] = 8

BABBAGE_FIRST: Final[int] = 7

BYRON: Final[int] = 1

BYRON_FIRST: Final[int] = 0

CONWAY: Final[int] = 11

CONWAY_FIRST: Final[int] = 9

DEFAULT_CLUSTER_ERA: Final[int] = 11

DEFAULT_TX_ERA: Final[int] = 11

DIJKSTRA: Final[int] = 12

DIJKSTRA_FIRST: Final[int] = 12

MAP: ClassVar[dict[int, EraName]] = {0: EraName.BYRON, 1: EraName.BYRON, 2: EraName.SHELLEY, 3: EraName.ALLEGRA, 4: EraName.MARY, 5: EraName.ALONZO, 6: EraName.ALONZO, 7: EraName.BABBAGE, 8: EraName.BABBAGE, 9: EraName.CONWAY, 10: EraName.CONWAY, 11: EraName.CONWAY, 12: EraName.DIJKSTRA}

MARY: Final[int] = 4

MARY_FIRST: Final[int] = 4

SHELLEY: Final[int] = 2

SHELLEY_FIRST: 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.

get_smash_version() → dict

Return version info for smash.

cardano_node_tests.utils.web module

Functionality for publishing content on an internal web server.

cardano_node_tests.utils.web.create_file(*, name: str, data: str, exist_ok: bool = False) → str

Create file on an internal web server and return the URL.

cardano_node_tests.utils.web.get_published_url(*, name: str = '') → str

Return the URL of the published file.

cardano_node_tests.utils.web.publish(*, file_path: Path, published_name: str = '', exist_ok: bool = False) → str

Publish file on an internal web server and return the URL.

cardano_node_tests.utils.web.unpublish(*, url: str, missing_ok: bool = True) → None

Unpublish file from the internal web server.

Module contents