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[source]

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[source]

Save CLI coverage info.

cardano_node_tests.utils.artifacts.save_cluster_artifacts(save_dir: Path, state_dir: Path) → None[source]

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

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

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 = '')[source]

Bases: object

Methods for working with GitHub issues.

issue[source]

A GitHub issue number.

repo[source]

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

fixed_in[source]

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

message[source]

A message to be added to blocking outcome.

copy() → GH[source]

Return a copy of the object.

finish_test() → None[source]

Fail or Xfail test with GitHub issue reference.

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

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)[source]

Bases: object

cluster_era: str[source]
command_era: str[source]
instance_num: int[source]
socket_path: Path[source]
state_dir: Path[source]
work_dir: Path[source]
class cardano_node_tests.utils.cluster_nodes.ClusterType[source]

Bases: object

Generic cluster type.

LOCAL: Final[str] = 'local'[source]
NODES: ClassVar[set[str]] = {}[source]
TESTNET: Final[str] = 'testnet'[source]
create_addrs_data(cluster_obj: ClusterLib, destination_dir: str | Path = '.') → dict[str, dict[str, Any]][source]

Create addresses and their keys for usage in tests.

get_cluster_obj(command_era: str = '') → ClusterLib[source]

Return instance of ClusterLib (cluster_obj).

test_addr_records: ClassVar[tuple[str, ...]] = ('user1', 'user2', 'user3', 'user4', 'user5')[source]
property testnet_type: str[source]
property uses_shortcut: bool[source]

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

class cardano_node_tests.utils.cluster_nodes.LocalCluster[source]

Bases: ClusterType

Local cluster type (full cardano mode).

NODES: ClassVar[set[str]] = {'bft1', 'pool1', 'pool2', 'pool3'}[source]
create_addrs_data(cluster_obj: ClusterLib, destination_dir: str | Path = '.') → dict[str, dict[str, Any]][source]

Create addresses and their keys for usage in tests.

get_cluster_obj(command_era: str = '') → ClusterLib[source]

Return instance of ClusterLib (cluster_obj).

property uses_shortcut: bool[source]

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

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

Bases: object

message: str = ''[source]
name: str[source]
pid: int | None[source]
status: str[source]
uptime: str | None[source]
class cardano_node_tests.utils.cluster_nodes.TestnetCluster[source]

Bases: ClusterType

Testnet cluster type (full cardano mode).

NODES: ClassVar[set[str]] = {'relay1'}[source]
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'}}[source]
create_addrs_data(cluster_obj: ClusterLib, destination_dir: str | Path = '.') → dict[str, dict[str, Any]][source]

Create addresses and their keys for usage in tests.

get_cluster_obj(command_era: str = '') → ClusterLib[source]

Return instance of ClusterLib (cluster_obj).

property testnet_type: str[source]

Return testnet type (shelley_qa, etc.).

property uses_shortcut: bool[source]

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

class cardano_node_tests.utils.cluster_nodes.Testnets[source]

Bases: object

mainnet: Final[str] = 'mainnet'[source]
preprod: Final[str] = 'preprod'[source]
preview: Final[str] = 'preview'[source]
cardano_node_tests.utils.cluster_nodes.get_cardano_node_socket_path(instance_num: int, socket_file_name: str = '') → Path[source]

Return path to socket file in the given cluster instance.

cardano_node_tests.utils.cluster_nodes.get_cluster_env() → ClusterEnv[source]

Get cardano cluster environment.

cardano_node_tests.utils.cluster_nodes.get_cluster_type() → ClusterType[source]

Return instance of the cluster type indicated by configuration.

cardano_node_tests.utils.cluster_nodes.get_instance_num() → int[source]

Get cardano cluster instance number.

cardano_node_tests.utils.cluster_nodes.load_addrs_data() → dict[source]

Load data about addresses and their keys for usage in tests.

cardano_node_tests.utils.cluster_nodes.load_pools_data(cluster_obj: ClusterLib) → dict[source]

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[source]

Reload supervisor configuration.

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

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[source]

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[source]

Run supervisorctl command.

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

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][source]

Return status info for list of services running on the running cluster (all by default).

cardano_node_tests.utils.cluster_nodes.set_cluster_env(instance_num: int, socket_file_name: str = '') → None[source]

Set env variables for the given cluster instance.

cardano_node_tests.utils.cluster_nodes.setup_test_addrs(cluster_obj: ClusterLib, destination_dir: str | Path = '.') → Path[source]

Set addresses and their keys up for usage in tests.

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

Start cluster.

cardano_node_tests.utils.cluster_nodes.start_nodes(node_names: list[str], instance_num: int | None = None) → None[source]

Start list of Cardano nodes of the running cluster.

cardano_node_tests.utils.cluster_nodes.stop_nodes(node_names: list[str], instance_num: int | None = None) → None[source]

Stop list of Cardano nodes of the running cluster.

cardano_node_tests.utils.cluster_scripts module

Functionality for cluster scripts (starting and stopping clusters).

  • copying scripts and their configuration, so it can be atered by tests

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

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

Bases: object

dir: Path[source]
start_script: Path[source]
start_script_args: list[str][source]
stop_script: Path[source]
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, ...])[source]

Bases: object

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

Bases: ScriptsTypes

Scripts for starting local cluster.

copy_scripts_files(destdir: str | Path) → StartupFiles[source]

Make copy of cluster scripts files located in this repository.

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

Generate topology files for split network.

get_instance_ports(instance_num: int) → InstancePorts[source]

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[source]

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)[source]

Bases: object

ekg: int[source]
node: int[source]
num: int[source]
prometheus: int[source]
class cardano_node_tests.utils.cluster_scripts.ScriptsTypes[source]

Bases: object

Generic cluster scripts.

LOCAL: Final[str] = 'local'[source]
TESTNET: Final[str] = 'testnet'[source]
copy_scripts_files(destdir: str | Path) → StartupFiles[source]

Make copy of cluster scripts files.

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

Generate topology files for split network.

get_instance_ports(instance_num: int) → InstancePorts[source]

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[source]

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)[source]

Bases: object

config_glob: str[source]
genesis_spec: Path[source]
start_script: Path[source]
class cardano_node_tests.utils.cluster_scripts.TestnetScripts[source]

Bases: ScriptsTypes

Scripts for starting a node on testnet.

BOOTSTRAP_CONF: Final[str] = 'testnet_conf'[source]
TESTNET_GLOBS: ClassVar[tuple[str, ...]] = ('config*.json', 'genesis-*.json', 'topology-*.json', 'dbsync-config.*', 'submit-api-config.*')[source]
copy_scripts_files(destdir: str | Path) → StartupFiles[source]

Make copy of cluster scripts files located in this repository.

get_instance_ports(instance_num: int) → InstancePorts[source]

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[source]

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)[source]

Bases: object

amount: int[source]
issuers_addrs: list[AddressRecord][source]
script: Path[source]
token: str[source]
token_mint_addr: AddressRecord[source]
class cardano_node_tests.utils.clusterlib_utils.TxMetadata(metadata: dict, aux_data: list)[source]

Bases: object

aux_data: list[source]
metadata: dict[source]
class cardano_node_tests.utils.clusterlib_utils.UpdateProposal(arg: str, value: Any, name: str = '', check_func: Callable | None = None)[source]

Bases: object

arg: str[source]
check_func: Callable | None = None[source]
name: str = ''[source]
value: Any[source]
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[source]

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[source]

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[source]

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[source]

Check that txins were spent.

cardano_node_tests.utils.clusterlib_utils.check_updated_params(update_proposals: list[UpdateProposal], protocol_params: dict) → None[source]

Compare update proposals with actual protocol parameters.

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

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

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

cardano_node_tests.utils.clusterlib_utils.create_collaterals(cluster: ClusterLib, payment_addr: AddressRecord, temp_template: str, tx_outs: list[TxOut]) → list[UTXOData][source]

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][source]

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][source]

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][source]

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[source]

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

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

Create new stake address(es).

cardano_node_tests.utils.clusterlib_utils.datum_hash_from_txout(cluster_obj: ClusterLib, txout: TxOut) → str[source]

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[source]

Deregister stake address.

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

Generate a Byron address and keys.

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

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

cardano_node_tests.utils.clusterlib_utils.get_delegation_state(cluster_obj: ClusterLib) → dict[source]

Get delegationState section of ledger state.

cardano_node_tests.utils.clusterlib_utils.get_ledger_state(cluster_obj: ClusterLib) → dict[source]

Return the current ledger state info.

cardano_node_tests.utils.clusterlib_utils.get_plutus_b64(script_file: str | Path) → str[source]

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][source]

Get cli arguments for pparams update action.

cardano_node_tests.utils.clusterlib_utils.get_snapshot_delegations(ledger_snapshot: dict) → dict[str, list[str]][source]

Get delegations data from ledger state snapshot.

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

Get uniform record for ledger state snapshot.

cardano_node_tests.utils.clusterlib_utils.get_utxo_ix_offset(utxos: list[UTXOData], txouts: list[TxOut]) → int[source]

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

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

cardano_node_tests.utils.clusterlib_utils.load_body_metadata(tx_body_file: Path) → Any[source]

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[source]

Load data of existing registered pool.

cardano_node_tests.utils.clusterlib_utils.load_tx_metadata(tx_body_file: Path) → TxMetadata[source]

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[source]

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

Positive amount value means minting, negative means burning.

cardano_node_tests.utils.clusterlib_utils.mint_or_burn_witness(cluster_obj: 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[source]

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

Positive amount value means minting, negative means burning.

cardano_node_tests.utils.clusterlib_utils.new_tokens(*asset_names: str, cluster_obj: ClusterLib, temp_template: str, token_mint_addr: AddressRecord, issuer_addr: AddressRecord, amount: int) → list[TokenRecord][source]

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[source]

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[source]

Save ledger state to file.

Parameters:

  • cluster_obj – An instance of clusterlib.ClusterLib.

  • state_name – A name of the ledger state (can be epoch number, etc.).

  • ledger_state – A dict with ledger state to save (optional).

  • destination_dir – A path to directory for storing the state JSON file (optional).

Returns:

A path to the generated state JSON file.

Return type:

Path

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

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[source]

Update params using update proposal.

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

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

Wait for time interval within an epoch.

Parameters:

  • cluster_obj – An instance of clusterlib.ClusterLib.

  • start – A start of the interval, in seconds. Negative number for counting from the end of an epoch.

  • stop – An end of the interval, in seconds. Negative number for counting from the end of an epoch.

  • force_epoch – A bool indicating whether the interval must be in current epoch (False by default).

  • check_slot – A bool indicating whether to check if slot number matches time interval after waiting (False by default).

cardano_node_tests.utils.clusterlib_utils.wait_for_rewards(cluster_obj: ClusterLib) → None[source]

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

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

Withdraw reward to payment address.

Parameters:

  • cluster_obj – An instance of clusterlib.ClusterLib.

  • stake_addr_record – An AddressRecord tuple for the stake address with reward.

  • dst_addr_record – An AddressRecord tuple for the destination payment address.

  • tx_name – A name of the transaction.

  • verify – A bool indicating whether to verify that the reward was transferred correctly.

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

cardano_node_tests.utils.configuration module

Cluster and test environment configuration.

cardano_node_tests.utils.dbsync_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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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[source]

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

cardano_node_tests.utils.dbsync_check_tx.utxodata2txout(utxodata: UTxORecord | UTXOData) → TxOut[source]

Convert UTxORecord or UTxOData to clusterlib.TxOut.

cardano_node_tests.utils.dbsync_check_tx.utxorecord2utxodata(utxorecord: UTxORecord) → UTXOData[source]

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[source]

Bases: object

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

conns: ClassVar[dict[int, connection | None]] = {0: None}[source]
cardano_node_tests.utils.dbsync_conn.close_all() → None[source]
cardano_node_tests.utils.dbsync_conn.conn() → connection[source]
cardano_node_tests.utils.dbsync_conn.reconn() → connection[source]

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)[source]

Bases: object

block_id: int[source]
deposits_drep: Decimal[source]
deposits_proposal: Decimal[source]
deposits_stake: Decimal[source]
epoch_no: int[source]
fees: Decimal[source]
id: int[source]
reserves: Decimal[source]
rewards: Decimal[source]
slot_no: int[source]
treasury: Decimal[source]
utxo: Decimal[source]
class cardano_node_tests.utils.dbsync_queries.ADAStashDBRow(id: int, addr_view: str, cert_index: int, amount: Decimal, tx_id: int)[source]

Bases: object

addr_view: str[source]
amount: Decimal[source]
cert_index: int[source]
id: int[source]
tx_id: int[source]
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)[source]

Bases: object

block_no: int | None[source]
epoch_no: int | None[source]
epoch_slot_no: int | None[source]
id: int[source]
pool_id: str | None[source]
previous_id: int | None[source]
proto_major: int | None[source]
proto_minor: int | None[source]
slot_no: int | None[source]
tx_count: int | None[source]
class cardano_node_tests.utils.dbsync_queries.CollateralTxOutDBRow(tx_out_id: int, utxo_ix: int, address: str, value: Decimal, tx_hash: memoryview)[source]

Bases: object

address: str[source]
tx_hash: memoryview[source]
tx_out_id: int[source]
utxo_ix: int[source]
value: Decimal[source]
class cardano_node_tests.utils.dbsync_queries.CommitteeDeregistrationDBRow(id: int, tx_id: int, cert_index: int, voting_anchor_id: int, cold_key: memoryview)[source]

Bases: object

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

Bases: object

cert_index: int[source]
cold_key: memoryview[source]
hot_key: memoryview[source]
id: int[source]
tx_id: int[source]
class cardano_node_tests.utils.dbsync_queries.DatumDBRow(id: int, datum_hash: memoryview, tx_id: int, value: dict, bytes: memoryview)[source]

Bases: object

bytes: memoryview[source]
datum_hash: memoryview[source]
id: int[source]
tx_id: int[source]
value: dict[source]
class cardano_node_tests.utils.dbsync_queries.DelegationVoteDBRow(id: int, stake_address_hash_view: str, drep_hash_view: str)[source]

Bases: object

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

Bases: object

amount: int[source]
drep_hash_view: str[source]
epoch_no: int[source]
hash_id: int[source]
id: int[source]
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)[source]

Bases: object

cert_index: int[source]
deposit: int[source]
drep_hash_id: int[source]
has_script: bool[source]
hash_raw: memoryview[source]
hash_view: str[source]
id: int[source]
tx_id: int[source]
voting_anchor_id: int | None[source]
class cardano_node_tests.utils.dbsync_queries.EpochDBRow(id: int, out_sum: int, fees: int, tx_count: int, blk_count: int, epoch_number: int)[source]

Bases: object

blk_count: int[source]
epoch_number: int[source]
fees: int[source]
id: int[source]
out_sum: int[source]
tx_count: int[source]
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)[source]

Bases: object

block_id: int[source]
coins_per_utxo_size: int[source]
collateral_percent: int[source]
committee_max_term_length: int[source]
committee_min_size: int[source]
cost_model_id: int[source]
decentralisation: float[source]
drep_activity: int[source]
drep_deposit: int[source]
dvt_committee_no_confidence: float[source]
dvt_committee_normal: float[source]
dvt_hard_fork_initiation: float[source]
dvt_motion_no_confidence: float[source]
dvt_p_p_economic_group: float[source]
dvt_p_p_gov_group: float[source]
dvt_p_p_network_group: float[source]
dvt_p_p_technical_group: float[source]
dvt_treasury_withdrawal: float[source]
dvt_update_to_constitution: float[source]
epoch_no: int[source]
extra_entropy: memoryview | None[source]
gov_action_deposit: int[source]
gov_action_lifetime: int[source]
id: int[source]
influence: float[source]
key_deposit: int[source]
max_bh_size: int[source]
max_block_ex_mem: int[source]
max_block_ex_steps: int[source]
max_block_size: int[source]
max_collateral_inputs: int[source]
max_epoch: int[source]
max_tx_ex_mem: int[source]
max_tx_ex_steps: int[source]
max_tx_size: int[source]
max_val_size: int[source]
min_fee_a: int[source]
min_fee_b: int[source]
min_fee_ref_script_cost_per_byte: float[source]
min_pool_cost: int[source]
min_utxo_value: int[source]
monetary_expand_rate: float[source]
nonce: memoryview[source]
optimal_pool_count: int[source]
pool_deposit: int[source]
price_mem: float[source]
price_step: float[source]
protocol_major: int[source]
protocol_minor: int[source]
pvt_committee_no_confidence: float[source]
pvt_committee_normal: float[source]
pvt_hard_fork_initiation: float[source]
pvt_motion_no_confidence: float[source]
pvtpp_security_group: float[source]
treasury_growth_rate: float[source]
class cardano_node_tests.utils.dbsync_queries.EpochStakeDBRow(id: int, hash: memoryview, view: str, amount: int, epoch_number: int)[source]

Bases: object

amount: int[source]
epoch_number: int[source]
hash: memoryview[source]
id: int[source]
view: str[source]
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)[source]

Bases: object

action_ix: int[source]
deposit: int[source]
description: dict[source]
dropped_epoch: int | None[source]
enacted_epoch: int | None[source]
expiration: int[source]
expired_epoch: int | None[source]
id: int[source]
param_proposal: int[source]
prev_gov_action_proposal: int | None[source]
ratified_epoch: int | None[source]
return_address: int[source]
tx_id: int[source]
type: str[source]
voting_anchor_id: int[source]
class cardano_node_tests.utils.dbsync_queries.MetadataDBRow(id: int, key: Decimal, json: Any, bytes: memoryview, tx_id: int)[source]

Bases: object

bytes: memoryview[source]
id: int[source]
json: Any[source]
key: Decimal[source]
tx_id: int[source]
class cardano_node_tests.utils.dbsync_queries.NewCommitteeInfoDBRow(id: int, tx_id: int, action_ix: int, quorum_numerator: int, quorum_denominator: int)[source]

Bases: object

action_ix: int[source]
id: int[source]
quorum_denominator: int[source]
quorum_numerator: int[source]
tx_id: int[source]
class cardano_node_tests.utils.dbsync_queries.NewCommitteeMemberDBRow(id: int, committee_id: int, committee_hash: memoryview, expiration_epoch: int)[source]

Bases: object

committee_hash: memoryview[source]
committee_id: int[source]
expiration_epoch: int[source]
id: int[source]
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)[source]

Bases: object

action_ix: int[source]
gap_id: int[source]
gov_action_type: str[source]
id: int[source]
script_hash: memoryview | None[source]
tx_id: int[source]
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)[source]

Bases: object

auth_name: str | None[source]
auth_pub_key: str | None[source]
auth_signature: str | None[source]
auth_warning: str | None[source]
auth_wit_alg: str | None[source]
data_bytes: memoryview[source]
data_comment: str | None[source]
data_hash: memoryview[source]
data_id: int[source]
data_is_valid: bool | None[source]
data_json: dict[source]
data_language: str[source]
data_vot_anchor_id: int[source]
data_warning: str | None[source]
ext_update_id: int | None[source]
ext_update_title: str | None[source]
ext_update_uri: str | None[source]
gov_act_abstract: str | None[source]
gov_act_id: int | None[source]
gov_act_motivation: str | None[source]
gov_act_rationale: str | None[source]
gov_act_title: str | None[source]
ref_hash_alg: str | None[source]
ref_hash_digest: str | None[source]
ref_id: int | None[source]
ref_label: str | None[source]
ref_uri: str | None[source]
vot_anchor_block_id: int[source]
vot_anchor_data_hash: memoryview[source]
vot_anchor_type: str[source]
vot_anchor_url: str[source]
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)[source]

Bases: object

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

Bases: object

fetch_error: str[source]
id: int[source]
voting_anchor_id: int[source]
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)[source]

Bases: object

coins_per_utxo_size: int | None[source]
collateral_percent: int | None[source]
committee_max_term_length: int | None[source]
committee_min_size: int | None[source]
cost_model_id: int | None[source]
decentralisation: float | None[source]
drep_activity: Decimal | None[source]
drep_deposit: int | None[source]
dvt_committee_no_confidence: float | None[source]
dvt_committee_normal: float | None[source]
dvt_hard_fork_initiation: float | None[source]
dvt_motion_no_confidence: float | None[source]
dvt_p_p_economic_group: float | None[source]
dvt_p_p_gov_group: float | None[source]
dvt_p_p_network_group: float | None[source]
dvt_p_p_technical_group: float | None[source]
dvt_treasury_withdrawal: float | None[source]
dvt_update_to_constitution: float | None[source]
entropy: memoryview | None[source]
epoch_no: int | None[source]
gov_action_deposit: int | None[source]
gov_action_lifetime: int | None[source]
id: int[source]
influence: float | None[source]
key: memoryview | None[source]
key_deposit: int | None[source]
max_bh_size: int | None[source]
max_block_ex_mem: int | None[source]
max_block_ex_steps: int | None[source]
max_block_size: int | None[source]
max_collateral_inputs: int | None[source]
max_epoch: int | None[source]
max_tx_ex_mem: int | None[source]
max_tx_ex_steps: int | None[source]
max_tx_size: int | None[source]
max_val_size: int | None[source]
min_fee_a: int | None[source]
min_fee_b: int | None[source]
min_fee_ref_script_cost_per_byte: float | None[source]
min_pool_cost: int | None[source]
min_utxo_value: int | None[source]
monetary_expand_rate: float | None[source]
optimal_pool_count: int | None[source]
pool_deposit: int | None[source]
price_mem: float | None[source]
price_step: float | None[source]
protocol_major: int | None[source]
protocol_minor: int | None[source]
pvt_committee_no_confidence: float | None[source]
pvt_committee_normal: float | None[source]
pvt_hard_fork_initiation: float | None[source]
pvt_motion_no_confidence: float | None[source]
pvtpp_security_group: float | None[source]
registered_tx_id: int | None[source]
treasury_growth_rate: float | None[source]
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)[source]

Bases: object

active_epoch_no: int[source]
cert_index: int[source]
deposit: Decimal | None[source]
dns_name: str | None[source]
fixed_cost: int[source]
hash: memoryview[source]
id: int[source]
ipv4: str | None[source]
ipv6: str | None[source]
margin: Decimal[source]
meta_id: int | None[source]
metadata_hash: memoryview | None[source]
metadata_url: str | None[source]
owner: memoryview[source]
owner_stake_address_id: int[source]
pledge: int[source]
port: int | None[source]
registered_tx_id: int[source]
retire_announced_tx_id: int | None[source]
retire_cert_index: int | None[source]
retiring_epoch: int | None[source]
reward_addr: memoryview[source]
reward_addr_view: str[source]
view: str[source]
vrf_key_hash: memoryview[source]
class cardano_node_tests.utils.dbsync_queries.PoolOffChainDataDBRow(id: int, ticker_name: str, hash: memoryview, json: dict, bytes: memoryview, pmr_id: int)[source]

Bases: object

bytes: memoryview[source]
hash: memoryview[source]
id: int[source]
json: dict[source]
pmr_id: int[source]
ticker_name: str[source]
class cardano_node_tests.utils.dbsync_queries.PoolOffChainFetchErrorDBRow(id: int, pmr_id: int, fetch_error: str, retry_count: int)[source]

Bases: object

fetch_error: str[source]
id: int[source]
pmr_id: int[source]
retry_count: int[source]
class cardano_node_tests.utils.dbsync_queries.PotTransferDBRow(id: int, cert_index: int, treasury: Decimal, reserves: Decimal, tx_id: int)[source]

Bases: object

cert_index: int[source]
id: int[source]
reserves: Decimal[source]
treasury: Decimal[source]
tx_id: int[source]
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)[source]

Bases: object

fee: int[source]
id: int[source]
purpose: str[source]
script_hash: memoryview[source]
tx_id: int[source]
unit_mem: int[source]
unit_steps: int[source]
value: dict[source]
class cardano_node_tests.utils.dbsync_queries.RewardDBRow(address: str, type: str, amount: Decimal, earned_epoch: int, spendable_epoch: int, pool_id: str | None = '')[source]

Bases: object

address: str[source]
amount: Decimal[source]
earned_epoch: int[source]
pool_id: str | None = ''[source]
spendable_epoch: int[source]
type: str[source]
class cardano_node_tests.utils.dbsync_queries.SchemaVersion[source]

Bases: object

Query and cache db-sync schema version.

classmethod stages() → SchemaVersionStages[source]
class cardano_node_tests.utils.dbsync_queries.SchemaVersionStages(one: int, two: int, three: int)[source]

Bases: object

one: int[source]
three: int[source]
two: int[source]
class cardano_node_tests.utils.dbsync_queries.ScriptDBRow(id: int, tx_id: int, hash: memoryview, type: str, serialised_size: int | None)[source]

Bases: object

hash: memoryview[source]
id: int[source]
serialised_size: int | None[source]
tx_id: int[source]
type: str[source]
class cardano_node_tests.utils.dbsync_queries.StakeAddrDBRow(id: int, view: str, tx_id: int)[source]

Bases: object

id: int[source]
tx_id: int[source]
view: str[source]
class cardano_node_tests.utils.dbsync_queries.StakeDelegDBRow(tx_id: int, active_epoch_no: int | None, pool_id: str | None, address: str | None)[source]

Bases: object

active_epoch_no: int | None[source]
address: str | None[source]
pool_id: str | None[source]
tx_id: int[source]
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)[source]

Bases: object

action_ix: int[source]
addr_view: str[source]
amount: Decimal[source]
dropped_epoch: int | None[source]
enacted_epoch: int | None[source]
expiration: int[source]
expired_epoch: int | None[source]
id: int[source]
ratified_epoch: int | None[source]
tx_id: int[source]
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)[source]

Bases: object

block_id: int[source]
block_index: int[source]
collateral_count: int[source]
collateral_out_count: int[source]
deposit: int[source]
extra_key_witness_count: int[source]
fee: Decimal[source]
invalid_before: Decimal | None[source]
invalid_hereafter: Decimal | None[source]
ma_tx_mint_id: int | None[source]
ma_tx_mint_name: memoryview | None[source]
ma_tx_mint_policy: memoryview | None[source]
ma_tx_mint_quantity: Decimal | None[source]
ma_tx_out_id: int | None[source]
ma_tx_out_name: memoryview | None[source]
ma_tx_out_policy: memoryview | None[source]
ma_tx_out_quantity: Decimal | None[source]
metadata_count: int[source]
out_sum: Decimal[source]
pot_transfer_count: int[source]
redeemer_count: int[source]
reference_input_count: int[source]
reserve_count: int[source]
script_count: int[source]
size: int[source]
stake_deleg_count: int[source]
stake_dereg_count: int[source]
stake_reg_count: int[source]
treasury_count: int[source]
treasury_donation: int[source]
tx_hash: memoryview[source]
tx_id: int[source]
tx_out_addr: str | None[source]
tx_out_addr_has_script: bool | None[source]
tx_out_data_hash: memoryview | None[source]
tx_out_id: int | None[source]
tx_out_inline_datum_hash: memoryview | None[source]
tx_out_reference_script_hash: memoryview | None[source]
tx_out_tx_id: int | None[source]
tx_out_value: Decimal | None[source]
utxo_ix: int | None[source]
withdrawal_count: int[source]
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)[source]

Bases: object

address: str[source]
ma_tx_out_id: int | None[source]
ma_tx_out_name: memoryview | None[source]
ma_tx_out_policy: memoryview | None[source]
ma_tx_out_quantity: Decimal | None[source]
reference_script_bytes: memoryview | None[source]
reference_script_hash: memoryview | None[source]
reference_script_json: dict | None[source]
reference_script_type: str | None[source]
tx_hash: memoryview[source]
tx_out_id: int[source]
utxo_ix: int[source]
value: Decimal[source]
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)[source]

Bases: object

address: str[source]
reference_script_bytes: memoryview | None[source]
reference_script_hash: memoryview | None[source]
reference_script_json: dict | None[source]
reference_script_type: str | None[source]
tx_hash: memoryview[source]
tx_out_id: int[source]
utxo_ix: int[source]
value: Decimal[source]
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)[source]

Bases: object

data_hash: memoryview | None[source]
has_script: bool[source]
payment_address: str[source]
stake_address: str | None[source]
tx_hash: memoryview[source]
utxo_ix: int[source]
value: int[source]
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)[source]

Bases: object

committee_voter: int | None[source]
drep_voter: int | None[source]
id: int[source]
pool_voter: int | None[source]
vote: str[source]
voter_role: str[source]
class cardano_node_tests.utils.dbsync_queries.WithdrawalDBRow(tx_id: int, address: str, amount: int)[source]

Bases: object

address: str[source]
amount: int[source]
tx_id: int[source]
cardano_node_tests.utils.dbsync_queries.execute(query: str, vars: Sequence = ()) → Iterator[cursor][source]
cardano_node_tests.utils.dbsync_queries.query_ada_pots(epoch_from: int = 0, epoch_to: int = 99999999) → Generator[ADAPotsDBRow, None, None][source]

Query ADA pots record in db-sync.

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

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][source]

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][source]

Query block records in db-sync.

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

Query transaction collateral txins in db-sync.

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

Query transaction collateral txouts in db-sync.

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

Query committee registration in db-sync.

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

Query committee members in db-sync.

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

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]][source]

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][source]

Query datum record in db-sync.

cardano_node_tests.utils.dbsync_queries.query_db_size() → int[source]

Query database size in MB in db-sync.

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

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][source]

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][source]

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][source]

Query epoch records in db-sync.

cardano_node_tests.utils.dbsync_queries.query_epoch_param(epoch_no: int = 0) → EpochParamDBRow[source]

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][source]

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][source]

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][source]

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][source]

Query new committee proposed in db-sync.

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

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][source]

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][source]

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][source]

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][source]

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][source]

Query off_chain_vote_fetch_error table in db-sync.

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

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][source]

Query pool data record in db-sync.

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

Query transaction redeemers in db-sync.

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

Query transaction reference txins in db-sync.

cardano_node_tests.utils.dbsync_queries.query_scripts(txhash: str) → Generator[ScriptDBRow, None, None][source]

Query transaction scripts in db-sync.

cardano_node_tests.utils.dbsync_queries.query_table_names() → list[str][source]

Query table names in db-sync.

cardano_node_tests.utils.dbsync_queries.query_treasury_withdrawal(txhash: str) → Generator[TreasuryWithdrawalDBRow, None, None][source]

Query treasury_withdrawal table in db-sync.

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

Query a transaction in db-sync.

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

Query transaction txins in db-sync.

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

Query transaction metadata in db-sync.

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

Query transaction MIR certificate records in db-sync.

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

Query transaction reserve record in db-sync.

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

Query stake registration record in db-sync.

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

Query stake deregistration record in db-sync.

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

Query stake registration record in db-sync.

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

Query transaction treasury record in db-sync.

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

Query reward withdrawal record in db-sync.

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

Query UTxOs for payment address in db-sync.

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

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)[source]

Bases: object

address: str[source]
amount: int[source]
cert_index: int[source]
class cardano_node_tests.utils.dbsync_types.CommitteeDeregistrationRecord(id: int, tx_id: int, cert_index: int, voting_anchor_id: int, cold_key: str)[source]

Bases: object

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

Bases: object

cert_index: int[source]
cold_key: str[source]
hot_key: str[source]
id: int[source]
tx_id: int[source]
class cardano_node_tests.utils.dbsync_types.DelegationRecord(address: str, pool_id: str, active_epoch_no: int)[source]

Bases: object

active_epoch_no: int[source]
address: str[source]
pool_id: str[source]
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)[source]

Bases: object

cert_index: int[source]
deposit: int[source]
drep_hash_id: int[source]
has_script: bool[source]
hash_bech32: str[source]
hash_hex: str[source]
id: int[source]
tx_id: int[source]
voting_anchor_id: int | None[source]
class cardano_node_tests.utils.dbsync_types.GetUTxORecord(utxo_hash: str, utxo_ix: int, has_script: bool, amount: int, data_hash: str)[source]

Bases: object

amount: int[source]
data_hash: str[source]
has_script: bool[source]
utxo_hash: str[source]
utxo_ix: int[source]
class cardano_node_tests.utils.dbsync_types.MetadataRecord(key: int, json: Any, bytes: memoryview)[source]

Bases: object

bytes: memoryview[source]
json: Any[source]
key: int[source]
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])[source]

Bases: object

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

Bases: object

amount_sum: int[source]
payment_address: str[source]
stake_address: str | None[source]
utxos: list[GetUTxORecord][source]
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)[source]

Bases: object

active_epoch_no: int[source]
cert_index: int[source]
fixed_cost: int[source]
hash: str[source]
id: int[source]
margin: float[source]
meta_id: int | None[source]
metadata_hash: str[source]
metadata_url: str[source]
owners: list[str][source]
pledge: int[source]
registered_tx_id: int[source]
relays: list[dict[str, dict[str, Any]]][source]
retire_announced_tx_id: int | None[source]
retire_cert_index: int | None[source]
retiring_epoch: int | None[source]
reward_addr: str[source]
view: str[source]
vrf_key_hash: str[source]
class cardano_node_tests.utils.dbsync_types.PotTransferRecord(treasury: int, reserves: int)[source]

Bases: object

reserves: int[source]
treasury: int[source]
class cardano_node_tests.utils.dbsync_types.RedeemerRecord(unit_mem: int, unit_steps: int, fee: int, purpose: str, script_hash: str, value: dict)[source]

Bases: object

fee: int[source]
purpose: str[source]
script_hash: str[source]
unit_mem: int[source]
unit_steps: int[source]
value: dict[source]
class cardano_node_tests.utils.dbsync_types.RewardEpochRecord(amount: int, earned_epoch: int, spendable_epoch: int, type: str, pool_id: str)[source]

Bases: object

amount: int[source]
earned_epoch: int[source]
pool_id: str[source]
spendable_epoch: int[source]
type: str[source]
class cardano_node_tests.utils.dbsync_types.RewardRecord(address: str, rewards: list[cardano_node_tests.utils.dbsync_types.RewardEpochRecord], reward_sum: int)[source]

Bases: object

address: str[source]
reward_sum: int[source]
rewards: list[RewardEpochRecord][source]
class cardano_node_tests.utils.dbsync_types.ScriptRecord(hash: str, type: str, serialised_size: int)[source]

Bases: object

hash: str[source]
serialised_size: int[source]
type: str[source]
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)[source]

Bases: object

last_row: TxDBRow[source]
ma_utxo_out: list[UTxORecord][source]
mint_utxo_out: list[UTxORecord][source]
utxo_out: list[UTxORecord][source]
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])[source]

Bases: object

block_id: int[source]
block_index: int[source]
collateral_outputs: list[UTXOData][source]
collaterals: list[UTxORecord][source]
deposit: int[source]
extra_key_witness: list[str][source]
fee: int[source]
invalid_before: int | None[source]
invalid_hereafter: int | None[source]
metadata: list[MetadataRecord][source]
mint: list[UTxORecord][source]
out_sum: int[source]</