cardano_node_tests.utils package

Submodules

cardano_node_tests.utils.artifacts module

Functionality for collecting testing artifacts.

cardano_node_tests.utils.artifacts.copy_artifacts(pytest_tmp_dir: Path, pytest_config: Config) → None

Copy collected tests and cluster artifacts to artifacts dir.

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.

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. Ignored on unknown projects.

message

A message to be added to blocking outcome.

copy() → GH

Return a copy of the object.

finish_test() → None

Fail or Xfail test with GitHub issue reference.

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

Fail or Xfail test with references to multiple GitHub issues.

cardano_node_tests.utils.cluster_nodes module

Functionality for cluster setup and interaction with cluster nodes.

class cardano_node_tests.utils.cluster_nodes.ClusterEnv(socket_path: pathlib.Path, state_dir: pathlib.Path, work_dir: pathlib.Path, instance_num: int, cluster_era: str, 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.ClusterType

Bases: object

Generic cluster type.

LOCAL: Final[str] = 'local'

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

TESTNET: Final[str] = 'testnet'

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

test_addr_records: ClassVar[tuple[str, ...]] = ('user1', 'user2', 'user3', 'user4', 'user5')

property testnet_type: str

property uses_shortcut: bool

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

class cardano_node_tests.utils.cluster_nodes.LocalCluster

Bases: ClusterType

Local cluster type (full cardano mode).

NODES: ClassVar[set[str]] = {'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 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[set[str]] = {'relay1'}

TESTNETS: ClassVar[dict[int, dict]] = {1506203091: {'shelley_start': '2020-07-29T21:44:51Z', 'type': 'mainnet'}, 1654041600: {'byron_epochs': 4, 'type': 'preprod'}, 1666656000: {'byron_epochs': 0, 'type': '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 (shelley_qa, etc.).

property uses_shortcut: bool

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

class cardano_node_tests.utils.cluster_nodes.Testnets

Bases: object

mainnet: Final[str] = 'mainnet'

preprod: Final[str] = 'preprod'

preview: Final[str] = 'preview'

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 atered by tests

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

class cardano_node_tests.utils.cluster_scripts.InstanceFiles(start_script: pathlib.Path, stop_script: pathlib.Path, start_script_args: list[str], dir: pathlib.Path)

Bases: object

dir: Path

start_script: Path

start_script_args: list[str]

stop_script: Path

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

Bases: object

base: int

bft1: int

ekg_bft1: int

ekg_pool1: int

ekg_pool2: int

ekg_pool3: int

ekg_relay1: int

metrics_submit_api: int

node_ports: tuple[NodePorts, ...]

pool1: int

pool2: int

pool3: int

prometheus_bft1: int

prometheus_pool1: int

prometheus_pool2: int

prometheus_pool3: int

prometheus_relay1: int

relay1: int

submit_api: int

supervisor: int

webserver: int

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

Bases: 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, start_script: str | Path = '', stop_script: str | Path = '') → InstanceFiles

Prepare scripts files for starting and stopping cluster instance.

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

Bases: object

ekg: int

node: int

num: int

prometheus: int

class cardano_node_tests.utils.cluster_scripts.ScriptsTypes

Bases: object

Generic cluster scripts.

LOCAL: Final[str] = 'local'

TESTNET: Final[str] = 'testnet'

copy_scripts_files(destdir: str | Path) → StartupFiles

Make copy of cluster scripts files.

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, start_script: str | Path = '', stop_script: str | Path = '') → 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', 'dbsync-config.*', 'submit-api-config.*')

copy_scripts_files(destdir: str | Path) → StartupFiles

Make copy of cluster scripts files located in this repository.

get_instance_ports(instance_num: int) → InstancePorts

Return ports mapping for given cluster instance.

prepare_scripts_files(destdir: str | Path, instance_num: int, start_script: str | Path = '', stop_script: str | Path = '') → InstanceFiles

Prepare scripts files for starting and stopping cluster instance.

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

cardano_node_tests.utils.clusterlib_utils module

Utilities that extends the functionality of cardano-clusterlib.

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

Bases: object

amount: int

issuers_addrs: list[AddressRecord]

script: Path

token: str

token_mint_addr: AddressRecord

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

Bases: object

aux_data: list

metadata: dict

class cardano_node_tests.utils.clusterlib_utils.UpdateProposal(arg: str, value: Any, name: str = '', check_func: 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 = '', use_build_cmd: bool = False, 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, 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 use_build_cmd to switch between transaction build and transaction build-raw. Use submit_method to switch between cardano-cli transaction submit and submit-api.

cardano_node_tests.utils.clusterlib_utils.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_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, 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, 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_script_context(cluster_obj: ClusterLib, plutus_version: int, redeemer_file: Path, tx_file: Path | None = None) → None

Run the create-script-context command (available in plutus-apps).

cardano_node_tests.utils.clusterlib_utils.create_stake_addr_records(*names: str, cluster_obj: 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.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_delegation_state(cluster_obj: ClusterLib) → dict

Get delegationState section of ledger state.

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_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_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[TokenRecord], temp_template: str, submit_method: str = 'cli', use_build_cmd: bool = False, 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[TokenRecord], temp_template: str, invalid_hereafter: int | None = None, invalid_before: int | None = None, submit_method: str = 'cli', use_build_cmd: bool = False, 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[TokenRecord]

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.update_params(cluster_obj: ClusterLib, src_addr_record: AddressRecord, update_proposals: list[UpdateProposal]) → None

Update params using update proposal.

cardano_node_tests.utils.clusterlib_utils.update_params_build(cluster_obj: ClusterLib, src_addr_record: AddressRecord, update_proposals: list[UpdateProposal]) → None

Update params using update proposal.

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

cardano_node_tests.utils.clusterlib_utils.wait_for_epoch_interval(cluster_obj: 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.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.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, 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

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.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_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_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_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_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_voting_procedure(txhash: str) → Generator[VotingProcedureDBRow, None, None]

Query voting_procedure table in db-sync.

cardano_node_tests.utils.dbsync_types module

Types used in db-sync related functions.

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

Bases: object

address: str

amount: int

cert_index: int

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

Bases: object

cert_index: int

cold_key: str

id: int

tx_id: int

voting_anchor_id: int

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

Bases: object

cert_index: int

cold_key: str

hot_key: str

id: int

tx_id: int

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

Bases: object

active_epoch_no: int

address: str

pool_id: str

class cardano_node_tests.utils.dbsync_types.DrepRegistrationRecord(id: int, tx_id: int, cert_index: int, deposit: int, drep_hash_id: int, voting_anchor_id: int | 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.

cardano_node_tests.utils.dbsync_utils.check_action_data(json_anchor_file: dict[str, Any], anchor_data_hash: str) → None

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

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

Check reward data for stake address in db-sync.

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

cardano_node_tests.utils.dbsync_utils.check_committee_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_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_param_proposal(protocol_params: dict) → ParamProposalDBRow | None

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

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.

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.

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

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

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

Check expected error on PoolOffChainFetchError.

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

Get off chain action data from db-sync.

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_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 AssertionError with NO_RESPONSE_STR until the expected data is returned.

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

Defragment address UTxOs.

cardano_node_tests.utils.faucet module

cardano_node_tests.utils.faucet.fund_from_faucet(*dst_addrs: AddressRecord | PoolUser, cluster_obj: ClusterLib, faucet_data: dict | None = None, all_faucets: dict[str, dict] | None = None, amount: int | list[int] | None = None, 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 | None

Get issue state.

property github: Github | None

is_closed() → bool

Check if issue is closed.

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(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

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(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

COMMITTEE = 'Committee'

CONSTITUTION = 'Constitution'

HARDFORK = 'HardFork'

PPARAM_UPDATE = 'PParamUpdate'

class cardano_node_tests.utils.governance_utils.ScriptTypes(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

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: list[list], 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_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(gov_state: dict[str, Any], action_txid: str, action_ix: int = 0) → dict[str, Any]

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

Wait until ratification is no longer delayed.

cardano_node_tests.utils.helpers module

cardano_node_tests.utils.helpers.callonce(func: TCallable) → TCallable

Call a function and cache its return value.

Warning

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

cardano_node_tests.utils.helpers.change_cwd(dir_path: str | Path) → Iterator[str | Path]

Change and restore CWD - context manager.

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_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, blocksize: int = 65536) → str

Return file checksum.

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

Convert from bech32 string.

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

Convert to bech32 string.

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

Temporarily set environment variables and restore previous environment afterwards.

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

Flatten an irregular (arbitrarily nested) iterable of iterables.

cardano_node_tests.utils.helpers.get_current_commit() → str

cardano_node_tests.utils.helpers.get_current_line_str() → str

Get filename#lineno of current line.

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

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

Return position of the current end of the file.

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

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

Return random string.

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

Return random string prefixed with timestamp.

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

cardano_node_tests.utils.helpers.get_vcs_link() → str

Return link to the current line in GitHub.

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

Ignore the KeyboardInterrupt signal.

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

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

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

Prepend flag to every item of the sequence.

Parameters:

• flag – A flag to prepend to every item of the contents.

• contents – A list (iterable) of content to be prepended.

Returns:

A list of flag followed by content, see below.

Return type:

list[str]

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

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

Run command.

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_arg(“create-script-context –plutus-v1”)

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

cardano_node_tests.utils.logfiles module

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

Bases: object

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.

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

Check if the expected messages are present in logs.

Parameters:

• regex_pairs (list[tuple[str, str]]) – List of tuples with file globs and regex patterns.

• seek_offsets (dict[str, int]) – Dictionary of file seek offsets.

• state_dir (Path) – Path to the state directory.

• timestamp (float) – Timestamp to filter log entries.

Returns:

List of error messages for missing log entries.

Return type:

list[str]

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

Cleanup relevant ignore rules file.

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

cardano_node_tests.utils.logfiles.expect_errors(regex_pairs: list[tuple[str, str]], worker_id: str) → Iterator[None]

Make sure the expected errors are present in logs.

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]]) → Iterator[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, only_first: bool = False) → list[str]

Find messages in log.

Parameters:

• regex (str) – The regular expression to search for.

• logfile (Path) – The path to the log file.

• seek_offset (int) – The seek offset in the log file.

• timestamp (float) – The timestamp to filter log entries.

• only_first (bool) – Whether to return only the first match.

Returns:

A list of matching log lines.

Return type:

list[str]

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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int1

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int2

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

node: Node

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

Bases: BaseModel

int: Int20

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

int: Int29

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

int: Int9

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int18

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

real: Real

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

server_timestamp_ms: ServerTimestampMs

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

Bases: BaseModel

int: Int3

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

node_is_leader: NodeIsLeader

node_not_leader: NodeNotLeader

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

Bases: BaseModel

int: Int8

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int6

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

mutator_cpu_ms: MutatorCpuMs

mutator_wall_ms: MutatorWallMs

num_bytes_usage_samples: NumBytesUsageSamples

num_gcs: NumGcs

par_avg_bytes_copied: ParAvgBytesCopied

par_max_bytes_copied: ParMaxBytesCopied

par_tot_bytes_copied: ParTotBytesCopied

peak_megabytes_allocated: PeakMegabytesAllocated

wall_ms: WallMs

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

int: Int23

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int22

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int24

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

int: Int26

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

counter: Counter

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: str

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

resident: Resident

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

Bases: BaseModel

int: Int21

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

myBlocksUncoupled: MyBlocksUncoupled

nodeIsLeaderNum: NodeIsLeaderNum

nodeStartTime: NodeStartTime

operationalCertificateExpiryKESPeriod: OperationalCertificateExpiryKESPeriod

operationalCertificateStartKESPeriod: OperationalCertificateStartKESPeriod

remainingKESPeriods: RemainingKESPeriods

served: Served

slotInEpoch: SlotInEpoch

slotNum: SlotNum

txsInMempool: TxsInMempool

txsProcessedNum: TxsProcessedNum

utxoSize: UtxoSize

class cardano_node_tests.utils.model_ekg.Model(*, iohk_monitoring_version: IohkMonitoringVersion, cardano: Cardano, ekg: Ekg, rts: Rts)

Bases: BaseModel

cardano: Cardano

ekg: Ekg

iohk_monitoring_version: IohkMonitoringVersion

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

rts: Rts

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

int: Int25

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int10

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

metrics: Metrics

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int4

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int5

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int12

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

int: Int15

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int14

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

mutticks: Mutticks

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: str

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

Bases: BaseModel

int: Int16

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int13

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

gc: Gc

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

header: Header

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

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

Bases: BaseModel

int: Int27

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int28

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

cputicks: Cputicks

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

threads: Threads

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

Bases: BaseModel

int: Int30

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int17

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int11

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

int: Int19

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

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

Bases: BaseModel

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

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

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

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

This replaces Model.__fields__ from Pydantic V1.

type: str

val: int

cardano_node_tests.utils.poll_utils module

Functions for working with SPO polls.

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

Bases: object

metadata: Path

poll: Path

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

Answer a poll and return the answer file.

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

Create a poll and return the poll and metadata files.

cardano_node_tests.utils.poll_utils.verify_poll(cluster_obj: ClusterLib, poll_file: Path, tx_signed: Path) → tuple[str, ...]

Verify an answer to the poll.

cardano_node_tests.utils.pytest_utils module

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

Bases: object

full: str

stage: str = ''

test_class: str = ''

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

Bases: object

CHANG_US: Final[str] = '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(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

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

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

Parameters:

• cluster_obj – An instance of clusterlib.ClusterLib.

• tx_file – A path to signed transaction file.

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

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

cardano_node_tests.utils.submit_api.submit_tx_bare(tx_file: 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

Bases: object

API: Final[str] = 'api'

CLI: Final[str] = 'cli'

cardano_node_tests.utils.submit_utils.is_submit_api_available() → bool

Check if cardano-submit-api is available.

cardano_node_tests.utils.submit_utils.submit_tx(submit_method: str, cluster_obj: 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.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 DRep certs.

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

Return a clusterlib.AddressRecord.

cardano_node_tests.utils.testnet_cleanup.deregister_stake_addr(cluster_obj: ClusterLib, pool_user: PoolUser, name_template: str, deposit_amt: int) → None

Deregister stake address.

cardano_node_tests.utils.testnet_cleanup.find_addr_files(location: Path) → Generator[Path, None, None]

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

cardano_node_tests.utils.testnet_cleanup.find_cert_files(location: Path) → Generator[Path, None, None]

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

cardano_node_tests.utils.testnet_cleanup.group_addr_files(file_paths: Generator[Path, None, None]) → list[list[Path]]

Group payment address files with corresponding stake address files.

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

cardano_node_tests.utils.testnet_cleanup.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, src_addrs: list[AddressRecord], faucet_address: str, tx_name: str) → None

Send funds from src_addr`s to `faucet_address.

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

Withdraw rewards to payment address.

cardano_node_tests.utils.tx_view module

Checks for transaction view CLI command.

cardano_node_tests.utils.tx_view.check_tx_view(cluster_obj: 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.Versions

Bases: object

Cluster era, transaction era, node version info.

ALLEGRA: Final[int] = 3

ALONZO: Final[int] = 6

BABBAGE: Final[int] = 8

BYRON: Final[int] = 1

CONWAY: Final[int] = 9

DEFAULT_CLUSTER_ERA: Final[int] = 9

DEFAULT_TX_ERA: Final[int] = 9

LAST_KNOWN_ERA: Final[int] = 9

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

MARY: Final[int] = 4

SHELLEY: Final[int] = 2

get_cardano_cli_version() → dict

Return version info for cardano-cli.

get_cardano_node_version() → dict

Return version info for cardano-node.

get_dbsync_version() → dict

Return version info for db-sync.

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