cardano_node_tests.tests package

Subpackages

Submodules

cardano_node_tests.tests.common module

cardano_node_tests.tests.common.allow_unstable_error_messages() → Iterator[None][source]

Catch AssertionError and either log it or raise it.

Used in tests where error messages can vary between node/CLI versions.

cardano_node_tests.tests.common.check_missing_utxos(cluster_obj: ClusterLib, utxos: list[UTXOData]) → None[source]: Fail if any node is missing the given UTxOs.

cardano_node_tests.tests.common.check_reference_script_policyid(name_template: str, cluster_obj: ClusterLib, script_file: str | Path, script_data: dict) → None[source]: Check that the policy ID of a reference script file matches the original script policy ID.

cardano_node_tests.tests.common.detect_fork(cluster_manager: ClusterManager, cluster_obj: ClusterLib, temp_template: str) → tuple[set[str], set[str]][source]: Detect if one or more nodes have forked blockchain or is out of sync.

cardano_node_tests.tests.common.fail_on_fork(cluster_manager: ClusterManager, cluster_obj: ClusterLib, temp_template: str) → None[source]: Fail if one or more nodes have forked blockchain or is out of sync.

cardano_node_tests.tests.common.get_conway_address_deposit(cluster_obj: ClusterLib) → int[source]: Get stake address deposit amount - is required in Conway+.

cardano_node_tests.tests.common.get_nodes_missing_utxos(cluster_obj: ClusterLib, utxos: list[UTXOData]) → set[str][source]: Return set of nodes that don’t have the given UTxOs.

cardano_node_tests.tests.common.get_payment_addr(name_template: str, cluster_manager: ClusterManager, cluster_obj: ClusterLib, caching_key: str = '', amount: int | None = None, min_amount: int | None = None, key_gen_method: KeyGenMethods = KeyGenMethods.DIRECT) → AddressRecord[source]: Create a single new payment address.

cardano_node_tests.tests.common.get_payment_addrs(name_template: str, cluster_manager: ClusterManager, cluster_obj: ClusterLib, num: int, fund_idx: list[int] | None = None, caching_key: str = '', amount: int | None = None, min_amount: int | None = None, key_gen_method: KeyGenMethods = KeyGenMethods.DIRECT) → list[AddressRecord][source]: Create new payment addresses.

cardano_node_tests.tests.common.get_pool_user(name_template: str, cluster_manager: ClusterManager, cluster_obj: ClusterLib, caching_key: str = '', amount: int | None = None, min_amount: int | None = None, payment_key_gen_method: KeyGenMethods = KeyGenMethods.DIRECT) → PoolUser[source]: Create a single new pool user.

cardano_node_tests.tests.common.get_pool_users(name_template: str, cluster_manager: ClusterManager, cluster_obj: ClusterLib, num: int, fund_idx: list[int] | None = None, caching_key: str = '', amount: int | None = None, min_amount: int | None = None, payment_key_gen_method: KeyGenMethods = KeyGenMethods.DIRECT) → list[PoolUser][source]: Create new pool users.

cardano_node_tests.tests.common.get_registered_pool_user(name_template: str, cluster_manager: ClusterManager, cluster_obj: ClusterLib, caching_key: str = '', amount: int | None = None, min_amount: int | None = None) → PoolUser[source]: Create new registered pool users.

cardano_node_tests.tests.common.get_test_id(cluster_or_manager: ClusterLib | ClusterManager) → str[source]

Return unique test ID - function name + assigned cluster instance + random string.

Log the test ID into cluster manager log file.

cardano_node_tests.tests.common.hypothesis_settings(max_examples: int = 100) → Any[source]

cardano_node_tests.tests.common.is_fee_in_interval(fee: float, expected_fee: float, frac: float = 0.1) → bool[source]

Check that the fee is within the expected range on local testnet.

The fee is considered to be within the expected range if it is within the expected_fee +/- frac range.

cardano_node_tests.tests.common.match_blocker(func: Callable) → Any[source]: Fail or Xfail the test if CLI error is raised.

cardano_node_tests.tests.common.unique_time_str() → str[source]

Return unique string based on current timestamp.

Useful for property-based tests as it isn’t possible to use random module in hypothesis tests.

cardano_node_tests.tests.conftest module

exception cardano_node_tests.tests.conftest.LogsError[source]: Bases: Exception

cardano_node_tests.tests.conftest.cd_testfile_temp_dir(testfile_temp_dir: Path) → Generator[Path, None, None][source]: Change to a temporary dir specific to a test file.

cardano_node_tests.tests.conftest.change_dir() → None[source]: Change CWD to temp directory before running tests.

cardano_node_tests.tests.conftest.close_dbconn() → Generator[None, None, None][source]: Close connection to db-sync database at the end of session.

cardano_node_tests.tests.conftest.cluster(cluster_manager: ClusterManager) → ClusterLib[source]: Return instance of clusterlib.ClusterLib.

cardano_node_tests.tests.conftest.cluster_lock_pool(cluster_manager: ClusterManager) → tuple[ClusterLib, str][source]: Lock any pool and return instance of clusterlib.ClusterLib.

cardano_node_tests.tests.conftest.cluster_manager(worker_id: str, request: FixtureRequest) → Generator[ClusterManager, None, None][source]: Return instance of cluster_management.ClusterManager.

cardano_node_tests.tests.conftest.cluster_singleton(cluster_manager: ClusterManager) → ClusterLib[source]: Lock whole cluster instance and return instance of clusterlib.ClusterLib.

cardano_node_tests.tests.conftest.cluster_use_pool(cluster_manager: ClusterManager) → tuple[ClusterLib, str][source]: Mark any pool as “in use” and return instance of clusterlib.ClusterLib.

cardano_node_tests.tests.conftest.function_autouse(cd_testfile_temp_dir: Generator[Path, None, None], respin_on_large_db: Generator[None, None, None]) → None[source]: Autouse function fixtures that are required for each test setup and teardown.

cardano_node_tests.tests.conftest.init_pytest_temp_dirs(tmp_path_factory: TempPathFactory) → None[source]: Init PytestTempDirs.

cardano_node_tests.tests.conftest.pytest_addoption(parser: Any) → None[source]

cardano_node_tests.tests.conftest.pytest_collection_modifyitems(config: Any, items: list) → None[source]

cardano_node_tests.tests.conftest.pytest_configure(config: Any) → None[source]

cardano_node_tests.tests.conftest.pytest_keyboard_interrupt() → None[source]: Create a status file indicating that the test run was interrupted.

cardano_node_tests.tests.conftest.respin_on_large_db(cluster_manager: ClusterManager) → Generator[None, None, None][source]: Respin a cluster instance running on CI when db-sync database size is over 256 MB.

cardano_node_tests.tests.conftest.session_autouse(init_pytest_temp_dirs: None, change_dir: None, close_dbconn: Any, testenv_setup_teardown: Any) → None[source]: Autouse session fixtures that are required for session setup and teardown.

cardano_node_tests.tests.conftest.submit_method(request: SubRequest) → str[source]: Return the submit method.

cardano_node_tests.tests.conftest.testenv_setup_teardown(worker_id: str, request: FixtureRequest) → Generator[None, None, None][source]: Setup and teardown test environment.

cardano_node_tests.tests.conftest.testfile_temp_dir() → Path[source]

Return a temporary dir for storing test artifacts.

The dir is specific to a single test file.

cardano_node_tests.tests.delegation module

Functionality for stake address delegation used in multiple tests modules.

class cardano_node_tests.tests.delegation.AddressRecordScript(address: str, script_file: pathlib._local.Path)[source]

Bases: object

address: str[source]

script_file: Path[source]

class cardano_node_tests.tests.delegation.DelegationOut(pool_user: cardano_clusterlib.structs.PoolUser, pool_id: str, tx_raw_output: cardano_clusterlib.structs.TxRawOutput)[source]

Bases: object

pool_id: str[source]

pool_user: PoolUser[source]

tx_raw_output: TxRawOutput[source]

class cardano_node_tests.tests.delegation.DelegationScriptOut(pool_user: cardano_node_tests.tests.delegation.PoolUserScript, pool_id: str, tx_raw_output: cardano_clusterlib.structs.TxRawOutput)[source]

Bases: object

pool_id: str[source]

pool_user: PoolUserScript[source]

tx_raw_output: TxRawOutput[source]

class cardano_node_tests.tests.delegation.PoolUserScript(payment: cardano_clusterlib.structs.AddressRecord, stake: cardano_node_tests.tests.delegation.AddressRecordScript)[source]

Bases: object

payment: AddressRecord[source]

stake: AddressRecordScript[source]

cardano_node_tests.tests.delegation.cluster_and_pool(cluster_manager: ClusterManager, use_resources: Iterable[str | BaseFilter] = ()) → tuple[ClusterLib, str][source]

Return instance of clusterlib.ClusterLib, and pool id to delegate to.

We need to mark the pool as “in use” when requesting local cluster instance, that’s why cluster instance and pool id are tied together in single fixture.

cardano_node_tests.tests.delegation.db_check_delegation(pool_user: PoolUser | PoolUserScript, db_record: TxRecord | None, deleg_epoch: int, pool_id: str, check_registration: bool = True)[source]: Check delegation in db-sync.

cardano_node_tests.tests.delegation.delegate_multisig_stake_addr(cluster_obj: ClusterLib, temp_template: str, pool_user: PoolUserScript, skey_files: Iterable[str | Path], pool_id: str = '', cold_vkey: Path | None = None, build_method=BuildMethods.BUILD_RAW) → DelegationScriptOut[source]: Submit registration certificate and delegate a multisig stake address to a pool.

cardano_node_tests.tests.delegation.delegate_stake_addr(cluster_obj: ClusterLib, addrs_data: dict, temp_template: str, pool_user: PoolUser | None = None, pool_id: str = '', cold_vkey: Path | None = None, amount: int = 100000000, build_method: str = BuildMethods.BUILD_RAW) → DelegationOut[source]: Submit registration certificate and delegate a stake address to a pool.

cardano_node_tests.tests.delegation.get_pool_id(cluster_obj: ClusterLib, addrs_data: dict, pool_name: str) → str[source]: Return stake pool id.

cardano_node_tests.tests.issues module

cardano_node_tests.tests.kes module

Functionality for KES key used in multiple tests modules.

class cardano_node_tests.tests.kes.KesScenarios[source]

Bases: object

ALL_INVALID = 'all_invalid'[source]

ALL_VALID = 'all_valid'[source]

INVALID_COUNTERS = 'invalid_counters'[source]

INVALID_KES_PERIOD = 'invalid_kes_period'[source]

cardano_node_tests.tests.kes.check_kes_period_info_result(cluster_obj: ClusterLib, kes_output: dict[str, Any], expected_scenario: str, check_id: str, expected_start_kes: int | None = None, pool_num: int | None = None) → list[str][source]

Check output kes-period-info command.

When pool_num is specified, prometheus metrics are checked.

cardano_node_tests.tests.kes.finish_on_errors(errors: list[str]) → None[source]: Fail or Xfail the test if there are errors.

cardano_node_tests.tests.kes.get_xfails(errors: list[str]) → list[GH][source]

Get xfail issues.

Either all errors can Xfail, or none of them can. There can be only one outcome of a test, so if there are errors that can’t be Xfailed, the test must fail.

cardano_node_tests.tests.plutus_common module

class cardano_node_tests.tests.plutus_common.ExecutionCost(per_time: int, per_space: int, fixed_cost: int)[source]

Bases: object

fixed_cost: int[source]

per_space: int[source]

per_time: int[source]

class cardano_node_tests.tests.plutus_common.PlutusOp(script_file: str | pathlib._local.Path, datum_file: pathlib._local.Path | None = None, datum_cbor_file: pathlib._local.Path | None = None, datum_value: str | None = None, redeemer_file: pathlib._local.Path | None = None, redeemer_cbor_file: pathlib._local.Path | None = None, redeemer_value: str | None = None, execution_cost: cardano_node_tests.tests.plutus_common.ExecutionCost | None = None)[source]

Bases: object

datum_cbor_file: Path | None = None[source]

datum_file: Path | None = None[source]

datum_value: str | None = None[source]

execution_cost: ExecutionCost | None = None[source]

redeemer_cbor_file: Path | None = None[source]

redeemer_file: Path | None = None[source]

redeemer_value: str | None = None[source]

script_file: str | Path[source]

class cardano_node_tests.tests.plutus_common.PlutusScriptData(script_file: pathlib._local.Path, script_type: str, execution_cost: cardano_node_tests.tests.plutus_common.ExecutionCost)[source]

Bases: object

execution_cost: ExecutionCost[source]

script_file: Path[source]

script_type: str[source]

class cardano_node_tests.tests.plutus_common.ScriptCost(fee: int, collateral: int, min_collateral: int)[source]

Bases: object

collateral: int[source]

fee: int[source]

min_collateral: int[source]

cardano_node_tests.tests.plutus_common.check_plutus_costs(plutus_costs: list[dict], expected_costs: list[ExecutionCost], frac: float = 0.15) → None[source]

Check plutus transaction cost.

units: the time is in picoseconds and the space is in bytes.

cardano_node_tests.tests.plutus_common.check_return_collateral(cluster_obj: ClusterLib, tx_output: TxRawOutput)[source]: Check if collateral is returned on Plutus script failure.

cardano_node_tests.tests.plutus_common.compute_cost(execution_cost: ExecutionCost, protocol_params: dict, collateral_fraction_offset: float = 1.0) → ScriptCost[source]: Compute fee and collateral required for the Plutus script.

cardano_node_tests.tests.plutus_common.create_script_context_w_blockers(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).

This variant of the create_script_context function catches known errors and fails the test with a blocker issue.

cardano_node_tests.tests.plutus_common.get_cost_per_unit(protocol_params: dict) → ExecutionCost[source]: Get execution cost per unit in Lovelace.

cardano_node_tests.tests.plutus_common.txout_factory(address: str, amount: int, plutus_op: PlutusOp, coin: str = 'lovelace', embed_datum: bool = False, inline_datum: bool = False) → TxOut[source]: Create TxOut object.

cardano_node_tests.tests.plutus_common.xfail_on_secp_error(cluster_obj: ClusterLib, algorithm: str, err_msg: str)[source]: Xfail a test based on error message when using SECP functions.

cardano_node_tests.tests.reqs_conway module

Conway User Stories.

cardano_node_tests.tests.test_addr_registration module

Tests for stake address registration.

class cardano_node_tests.tests.test_addr_registration.TestNegative[source]

Bases: object

Tests that are expected to fail.

test_deregister_not_registered_addr(cluster: ClusterLib, pool_users: list[PoolUser], pool_users_disposable: list[PoolUser], build_method: str)[source]

Deregister not registered stake address.

Uses parametrized build method (build, build-raw, or build-estimate).

Expect failure.

Create stake address but do not register it
Generate stake address deregistration certificate for unregistered address
Attempt to build/submit transaction to deregister unregistered stake address
Check that transaction fails with StakeKeyNotRegisteredDELEG error

test_incomplete_multisig(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str, issue: str)[source]

Try to register a multisig stake address while missing either a script or an skey.

Expect failure.

Create a multisig script to be used as stake credentials
Create stake address registration certificate
Create a Tx for the registration certificate
Scenario1: Build a Tx with the registration certificate, without passing the the script to the transaction build command
Scenario 2: One skey is missing when witnesing the Tx
Incrementally sign the Tx and submit the registration certificate
Check the expected failure

test_register_addr_with_wrong_key(cluster: ClusterLib, pool_users: list[PoolUser], pool_users_disposable: list[PoolUser])[source]

Try to register stake address using wrong payment skey.

Expect failure.

Create stake address registration certificate
Attempt to submit registration transaction signed with wrong payment skey
Check that transaction submission fails with MissingVKeyWitnessesUTXOW error

test_registration_cert_with_wrong_key(cluster: ClusterLib, pool_users: list[PoolUser])[source]

Try to generate stake address registration certificate using wrong stake vkey.

Expect failure.

Attempt to generate stake address registration certificate using payment vkey instead of stake vkey
Check that certificate generation fails with expected error about wrong key type

class cardano_node_tests.tests.test_addr_registration.TestRegisterAddr[source]

Bases: object

Tests for stake address registration and deregistration.

test_addr_registration_certificate_order(cluster: ClusterLib, pool_users: list[PoolUser], pool_users_disposable: list[PoolUser], submit_method: str, build_method: str)[source]

Submit (de)registration certificates in single TX and check that the order matter.

Create stake address registration cert
Create stake address deregistration cert
Register, deregister, register, deregister and register stake address in single TX
Check that the address is registered
Check that the balance for source address was correctly updated and that key deposit was needed
(optional) Check records in db-sync

test_addr_registration_deregistration(cluster: ClusterLib, pool_users: list[PoolUser], pool_users_disposable: list[PoolUser], build_method: str)[source]

Submit registration and deregistration certificates in single TX.

Create stake address registration cert
Create stake address deregistration cert
Register and deregister stake address in single TX
Check that the balance for source address was correctly updated and that key deposit was not needed
(optional) Check records in db-sync

test_deregister_registered(cluster: ClusterLib, pool_users: list[PoolUser], pool_users_disposable: list[PoolUser], build_method: str)[source]

Deregister a registered stake address.

Create stake address registration cert
Register and stake address
Create stake address deregistration cert
Deregister stake address
Check that the balance for source address was correctly updated
(optional) Check records in db-sync

test_multisig_deregister_registered(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str, key_type: str)[source]

Deregister a registered multisig stake address.

Create a multisig script to be used as stake credentials
Create stake address registration certificate
Create a Tx for the registration certificate
Incrementally sign the Tx and submit the registration certificate
Check that the address is registered
Create stake address deregistration certificate
Create a Tx for the deregistration certificate
Incrementally sign the Tx and submit the deregistration certificate
Check that the address is no longer registered
(optional) Check records in db-sync

cardano_node_tests.tests.test_addr_registration.pool_users(cluster_manager: ClusterManager, cluster: ClusterLib) → list[PoolUser][source]: Create pool users.

cardano_node_tests.tests.test_addr_registration.pool_users_disposable(cluster: ClusterLib) → list[PoolUser][source]: Create function scoped pool users.

cardano_node_tests.tests.test_blocks module

Tests for blocks production.

Other block production checks may be present in test_staking.py.

class cardano_node_tests.tests.test_blocks.TestCollectData[source]

Bases: object

Tests for collecting data about blocks production.

block_production_db() → Generator[Connection, None, None][source]: Open block production db.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

pytestmark = [Mark(name='skipif', args=(True,), kwargs={'reason': 'runs only during block-production testing'})][source]

test_block_production(cluster_manager: ClusterManager, cluster: ClusterLib, payment_addrs: list[AddressRecord], block_production_db: Connection)[source]

Record number of blocks produced by each pool over multiple epochs.

Register and delegate a stake address to every pool
Collect block production data over multiple epochs
- Each epoch save ledger state
- Each epoch save block production data to sqlite db
- Transfer funds between multiple addresses to create activity

class cardano_node_tests.tests.test_blocks.TestDynamicBlockProd[source]

Bases: object

Tests for P2P dynamic block production.

backup_node_name = 'pool2'[source]

payment_addrs(cluster_manager: ClusterManager, cluster_singleton: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

producing_node_name = 'pool1'[source]

reconf_for_dynamic() → None[source]

Reconfigure cluster for dynamic block production.

Reconfigure nodeX to be backup block production node for nodeY.

test_dynamic_block_production(cluster_manager: ClusterManager, cluster_singleton: ClusterLib, payment_addrs: list[AddressRecord])[source]

Check dynamic block production.

Check that blocks are produced by both nodeX and nodeY pools
Reconfigure a nodeX to be non-producing backup node for nodeY pool
Terminate nodeY and send SIGHUP to nodeX so it starts producing blocks on behalf of nodeY
Check that nodeX has replaced nodeY and is producing blocks on its behalf

class cardano_node_tests.tests.test_blocks.TestLeadershipSchedule[source]

Bases: object

Tests for cardano-cli leadership-schedule.

test_pool_blocks(cluster_manager: ClusterManager, cluster_use_pool: tuple[ClusterLib, str], for_epoch: str)[source]

Check that blocks were forged according to leadership schedule.

Query leadership schedule for selected pool for current epoch or next epoch
Wait for epoch that comes after the queried epoch
Get info about forged blocks in queried epoch for the selected pool
Compare leadership schedule with blocks that were actually forged
Compare log records with ledger state dump
(optional) Check forged blocks in db-sync

test_unstable_stake_distribution(cluster_manager: ClusterManager, cluster: ClusterLib)[source]

Try to query leadership schedule for next epoch when stake distribution is unstable.

Expect failure.

Test that querying leadership schedule fails when stake distribution is not yet stable for next epoch. Stable stake distribution is only available in last 300 slots of current epoch.

Wait for epoch interval where stake distribution for next epoch is unstable (before last 300 slots of current epoch)
Attempt to query leadership schedule for next epoch using pool VRF and cold keys
Check that query fails with error message about unstable stake distribution

cardano_node_tests.tests.test_chain_transactions module

Tests for transactions chaining.

class cardano_node_tests.tests.test_chain_transactions.TestTxChaining[source]

Bases: object

cluster(cluster_manager: ClusterManager) → ClusterLib[source]

test_tx_chaining(cluster_manager: ClusterManager, cluster: ClusterLib)[source]

Test transaction chaining with 1000 dependent transactions in mempool.

Submit Txs one by one without waiting for them to appear on ledger and use output of one Tx as input for the next Tx. As a result, Txs are using inputs that don’t exist on ledger yet. Node needs to lookup the Tx in mempool and correctly chain the dependent transactions.

Create payment address with 205 ADA
Generate 1000 signed transactions where each uses output of previous transaction
Submit all transactions to mempool without waiting for ledger confirmation
Verify node can chain dependent transactions through mempool lookups
Check balances after all transactions are processed
(optional) Check transactions in db-sync

cardano_node_tests.tests.test_cli module

Tests for cardano-cli that doesn’t fit into any other test file.

class cardano_node_tests.tests.test_cli.TestAddressBuild[source]

Bases: object

Tests for cardano-cli address build.

test_address_build(cluster: ClusterLib, payment: str, stake: str)[source]

Build payment addresses with different input combinations.

Test parametrized across payment credential types (vkey, vkey_file, script_file) and stake credential types (None, vkey, vkey_file, script_file, address).

Load golden payment and stake verification keys and script files
Build payment address using specified payment credential type (parameter)
Optionally include stake credential using specified type (parameter)
Verify generated address matches one of expected address formats
Check address type varies based on stake credential (payment-only or base address)

test_invalid_payment_info(cluster: ClusterLib, option: str) → None[source]

Try to use ‘address build’ with invalid payment information (property-based test).

Expect failure.

test_invalid_stake_info(cluster: ClusterLib, option: str) → None[source]

Try to use ‘address build’ with invalid stake address information (property-based test).

Expect failure.

class cardano_node_tests.tests.test_cli.TestAddressInfo[source]

Bases: object

Tests for cardano-cli address info.

test_address_info_payment(cluster: ClusterLib, addr_gen: str)[source]

Retrieve and verify payment address information.

Test parametrized with static and dynamically generated payment addresses.

Use static payment address or generate new payment address (addr_gen parameter)
Execute cardano-cli address info command
Verify address field matches input address
Verify era is “shelley”
Verify encoding is “bech32”
Verify type is “payment”
For static address, verify base16 encoding matches expected value

test_address_info_payment_with_outfile(cluster: ClusterLib)[source]

Compare payment address info with and without outfile provided.

Test that address info produces identical output when written to stdout via –out-file compared to default stdout output.

Execute cardano-cli address info without –out-file argument
Parse JSON output from stdout
Execute cardano-cli address info –out-file /dev/stdout
Parse JSON output from stdout with –out-file
Verify both outputs are identical

test_address_info_script(cluster: ClusterLib)[source]

Retrieve and verify script address information.

Test that address info correctly identifies script addresses.

Generate payment address and keys
Create multisig script using payment verification key
Generate script address from multisig script
Execute cardano-cli address info command for script address
Verify address field matches input address
Verify era is “shelley”
Verify encoding is “bech32”
Verify type is “payment” (script addresses have payment type)

test_address_info_stake(cluster: ClusterLib, addr_gen: str)[source]

Retrieve and verify stake address information.

Test parametrized with static and dynamically generated stake addresses.

Use static stake address or generate new stake address (addr_gen parameter)
Execute cardano-cli address info command
Verify address field matches input address
Verify era is “shelley”
Verify encoding is “bech32”
Verify type is “stake”
For static address, verify base16 encoding matches expected value

test_address_info_with_invalid_address(cluster: ClusterLib) → None[source]

Try to use ‘address info’ with invalid address (property-based test).

Expect failure.

class cardano_node_tests.tests.test_cli.TestAddressKeyHash[source]

Bases: object

Tests for cardano-cli address key-hash.

test_invalid_verification_key(cluster: ClusterLib, option: str) → None[source]

Try to use address key-hash with invalid verification key (property-based test).

Expect failure.

test_valid_verification_key(cluster: ClusterLib, option: str)[source]

Compute payment verification key hash from valid key.

Test parametrized with vkey (bech32 string) and vkey_file (file path) input options.

Load golden payment verification key file
Convert to bech32 format if option is “vkey”
Execute cardano-cli address key-hash command with specified input option
Verify computed hash matches expected golden hash value

class cardano_node_tests.tests.test_cli.TestAdvancedQueries[source]

Bases: object

Basic sanity tests for advanced cardano-cli query commands.

The query leadership-schedule is handled by more complex tests TestLeadershipSchedule as it requires complex setup. For query protocol-state see test_protocol_state_keys smoke test.

pool_ids(cluster: ClusterLib) → list[str][source]

test_ledger_state(cluster: ClusterLib)[source]

Query ledger state and validate structure.

Test that ledger-state query returns expected JSON structure containing epoch information.

Execute cardano-cli query ledger-state command
Verify response contains expected “lastEpoch” field
Check that ledger state JSON can be parsed successfully
Handle known node issue 3859 for invalid numeric literals in JSON output

test_pool_params(cluster: ClusterLib, pool_ids: list[str])[source]

Query pool parameters and validate structure.

Test that pool-params query returns expected pool configuration fields.

Get first available pool ID from cluster
Execute cardano-cli query pool-params command for pool
Verify response contains expected “retiring” field
Check that pool parameters JSON can be parsed successfully

test_pool_state(cluster: ClusterLib, pool_ids: list[str])[source]

Query pool state and validate structure.

Test that pool-state query returns expected pool runtime state fields.

Get first available pool ID from cluster
Execute cardano-cli query pool-state command for pool
Verify response contains expected “retiring” field
Check that pool state can be retrieved successfully

test_stake_snapshot(cluster_manager: ClusterManager, cluster: ClusterLib, option: str)[source]

Query stake snapshot for pools and validate structure.

Test parametrized to query single pool, multiple pools, or all pools. See also TestLedgerState.test_stake_snapshot for comprehensive ledger state validation.

Get stake pool IDs from cluster
Execute cardano-cli query stake-snapshot with specified pool filter (option parameter)
Verify response contains expected snapshot fields (stakeMark, stakeSet, stakeGo)
Check pool stake totals match expected values
(optional) Validate stake distribution against db-sync for single_pool option

test_tx_mempool_info(cluster: ClusterLib, with_out_file: bool)[source]

Test ‘query tx-mempool info’.

Check that the expected fields are returned
Check that the slot number returned is the last applied on the ledger plus one

class cardano_node_tests.tests.test_cli.TestCLI[source]

Bases: object

Tests for cardano-cli.

TX_BODY_FILE = PosixPath('/home/martink/Source/repos/cardano-node-tests/cardano_node_tests/tests/data/test_tx_metadata_both_tx.body')[source]

TX_BODY_FILE_CONWAY = PosixPath('/home/martink/Source/repos/cardano-node-tests/cardano_node_tests/tests/data/test_tx_metadata_both_tx_conway.body')[source]

TX_BODY_OUT_JSON = PosixPath('/home/martink/Source/repos/cardano-node-tests/cardano_node_tests/tests/data/test_tx_metadata_both_tx_body_json.out')[source]

TX_FILE = PosixPath('/home/martink/Source/repos/cardano-node-tests/cardano_node_tests/tests/data/test_tx_metadata_both_tx.signed')[source]

TX_OUT_JSON = PosixPath('/home/martink/Source/repos/cardano-node-tests/cardano_node_tests/tests/data/test_tx_metadata_both_tx_json.out')[source]

test_calculate_min_fee(cluster: ClusterLib)[source]

Calculate minimum transaction fee using cardano-cli.

Test that transaction calculate-min-fee command returns reasonable fee value in Lovelace.

Create protocol parameters file
Execute cardano-cli transaction calculate-min-fee command
Provide transaction body file, witness count, input/output counts
Check that calculated fee is below maximum expected threshold (172000 Lovelace)
Verify output format includes fee value and “Lovelace” unit

test_ledger_peer_snapshot(cluster: ClusterLib)[source]

Query ledger peer snapshot and validate structure.

Test that ledger peer snapshot command returns expected JSON structure with pool stake information.

Execute cardano-cli query ledger-peer-snapshot command
Verify response contains expected keys (bigLedgerPools, slotNo, version)
Check that bigLedgerPools is a list
Verify each pool entry contains relativeStake field if pools exist

test_protocol_mode(cluster: ClusterLib)[source]

Check the default protocol mode - command works even without specifying protocol mode.

Test that cardano-cli query commands work without explicitly specifying the protocol mode.

Execute cardano-cli query utxo command without –cardano-mode or –byron-mode flag
Use static testnet address for query
Verify command executes successfully without protocol mode argument
Check that default protocol mode is automatically applied

test_query_tip(cluster: ClusterLib)[source]

Query blockchain tip and validate output fields.

Test that query tip command returns expected fields with correct values and constraints.

Execute cardano-cli query tip command
Check that output contains expected fields (block, epoch, era, hash, slot, syncProgress)
Check for optional fields slotInEpoch and slotsToEpochEnd (node 1.36.0+)
Verify slotInEpoch is never greater than epoch length
Verify slotsToEpochEnd matches difference between epoch length and slotInEpoch
Verify slot number is within valid range for current epoch
Verify era matches expected cluster era name

test_sign_tx_with_process_substitution(cluster: ClusterLib)[source]

Sign transaction using signing key file via process substitution.

Test that cardano-cli accepts signing key file content via bash process substitution instead of file path.

Load signing key file content into bash variable
Execute cardano-cli transaction sign with process substitution for –signing-key-file
Verify command accepts process substitution syntax for key file
Check that transaction is signed successfully

test_toplevel_queries(cluster: ClusterLib, command: str)[source]

Verify that query commands are accessible from top level cardano-cli.

Test parametrized across multiple query commands to ensure they are available in top-level command structure rather than requiring era-specific paths.

Execute cardano-cli query <command> without era prefix
Verify command is recognized (may fail with usage error if args missing)
Check that command does not produce “Invalid argument ‘query’” error
Verify usage message contains expected command name

test_tx_view(cluster: ClusterLib)[source]

View transaction body and signed transaction in JSON format.

Test that transaction view command produces expected JSON output for both transaction body and signed transaction files.

Execute cardano-cli transaction view –tx-body-file command
Check output includes redeemers and datums fields
Compare transaction body JSON output with golden file
Execute cardano-cli transaction view –tx-file command for signed transaction
Compare signed transaction JSON output with golden file

test_txid_with_process_substitution(cluster: ClusterLib)[source]

Extract transaction ID from transaction file using process substitution.

Test that cardano-cli accepts transaction file content via bash process substitution instead of file path.

Load transaction file content into bash variable
Execute cardano-cli transaction txid with process substitution <(echo content)
Verify command accepts process substitution syntax
Check that transaction ID is extracted successfully

class cardano_node_tests.tests.test_cli.TestKey[source]

Bases: object

Tests for cardano-cli key.

test_non_extended_key_error(cluster: ClusterLib)[source]

Try to get a non-extended verification key with a signing key file.

Expect failure. Should only allow extended verification key files.

test_non_extended_key_valid(cluster: ClusterLib)[source]

Derive non-extended verification key from extended verification key.

Test that non-extended key derivation produces valid key that is prefix of extended key.

Generate extended payment key pair
Extract extended verification key CBOR hex (excluding first 4 chars)
Execute cardano-cli key non-extended-key command with extended verification key file
Extract non-extended verification key CBOR hex
Verify extended key starts with non-extended key (non-extended is 32-byte prefix)

test_stake_non_extended_key(cluster: ClusterLib)[source]

Derive stake non-extended key from stake extended key.

Test that stake non-extended key derivation works correctly with stake extended keys.

Load golden stake extended verification key file
Execute cardano-cli key non-extended-key command with stake extended key file
Verify command succeeds (or handle known issue node_4914 if error occurs)
Check that non-extended stake key is generated successfully

class cardano_node_tests.tests.test_cli.TestPing[source]

Bases: object

Tests for cardano-cli ping.

test_ping_localhost(cluster: ClusterLib)[source]

Ping local node using TCP host and port.

Test that cardano-cli ping command successfully connects to local node via TCP.

Get pool1 or relay1 port from cluster instance configuration
Execute cardano-cli ping command with –host localhost and –port
Send 5 ping requests with JSON output
Verify JSON response contains pongs array
Check that last pong has cookie value matching request count minus 1 (4)

test_ping_tip(cluster: ClusterLib)[source]

Query blockchain tip using cardano-cli ping command.

Test that ping –tip flag returns current blockchain tip information.

Get pool1 or relay1 port from cluster instance configuration
Execute cardano-cli ping –tip command with –host and –port
Parse JSON output and extract slotNo from last tip entry
Query current tip using standard query tip command
Verify ping tip slotNo is within 100 slots of cluster tip

test_ping_unix_socket(cluster: ClusterLib)[source]

Ping local node using Unix domain socket.

Test that cardano-cli ping command successfully connects to local node via Unix socket.

Add log ignore rule for expected MuxUnknownMiniProtocol errors (ping protocol)
Execute cardano-cli ping command with –unixsock and CARDANO_NODE_SOCKET_PATH
Send 5 ping requests with JSON output
Handle expected “Unix sockets only support queries” error on newer CLI versions
Verify JSON response contains pongs array
Check that last pong has cookie value matching request count minus 1 (4)

test_ping_version(cluster: ClusterLib)[source]

Query node-to-node protocol version using cardano-cli ping.

Test that ping –query-versions returns protocol version information.

Get pool1 or relay1 port from cluster instance configuration
Execute cardano-cli ping –query-versions command with –host and –port
Handle expected UnknownVersionInRsp error when CLI and node versions mismatch
Verify output contains “NodeToNodeVersion” when versions match
Check for known issue network_5281 if versions match but error occurs

class cardano_node_tests.tests.test_cli.TestQuerySlotNumber[source]

Bases: object

Tests for cardano-cli query slot-number.

test_slot_number(cluster: ClusterLib)[source]

Convert UTC timestamp to slot number.

Test that slot-number query correctly converts current UTC timestamp to slot number.

Get current UTC timestamp
Execute cardano-cli query slot-number command with timestamp
Wait for new block to ensure tip is up-to-date
Query current blockchain tip
Verify slot number is within valid range (not greater than total slots for epoch)

test_slot_number_invalid_format(cluster: ClusterLib)[source]

Test query slot-number with a timestamp invalid format.

Expect failure.

test_slot_number_out_of_range(cluster: ClusterLib, time_val: str)[source]

Test query slot-number with a timestamp out of range.

Expect failure.

class cardano_node_tests.tests.test_cli.TestQueryUTxO[source]

Bases: object

Tests for cardano-cli query utxo.

test_address_invalid_data(cluster: ClusterLib) → None[source]

Try to use ‘query utxo’ with invalid ‘address’ (property-based test).

Expect failure.

test_pretty_utxo(cluster_manager: ClusterManager, cluster: ClusterLib)[source]

Verify pretty-printed UTxO query output format.

Test that text-formatted UTxO output contains expected fields in correct order and format.

Create source and destination payment addresses
Fund source address with 14 ADA (2 + 2.5 + 10 for fees)
Send 2 ADA and 2.5 ADA to destination address in single transaction
Execute cardano-cli query utxo –output-text for destination address
Verify output contains expected columns (TxHash, TxIx, Amount)
Check transaction ID, indices (0, 1), amounts, and TxOutDatumNone match expectations

test_tx_in_invalid_data(cluster: ClusterLib, invalid_param: str) → None[source]

Try to use ‘query utxo’ with invalid ‘tx-in’ (property-based test).

Expect failure.

test_whole_utxo(cluster: ClusterLib)[source]

Query entire UTxO set from blockchain.

Test that whole-utxo query returns complete UTxO set on local cluster without errors.

Execute cardano-cli query utxo –whole-utxo command
Verify command completes successfully
Check that entire UTxO set can be retrieved (may be large on mainnet)

class cardano_node_tests.tests.test_cli.TestStakeAddressKeyHash[source]

Bases: object

Tests for cardano-cli stake-address key-hash.

test_invalid_verification_key(cluster: ClusterLib, option: str) → None[source]

Try to use stake-address key-hash with invalid verification key (property-based test).

Expect failure.

test_valid_verification_key(cluster: ClusterLib, option: str)[source]

Compute stake verification key hash from valid key.

Test parametrized with vkey (bech32 string) and vkey_file (file path) input options.

Load golden stake verification key file
Convert to bech32 format if option is “vkey”
Execute cardano-cli stake-address key-hash command with specified input option
Verify computed hash matches expected golden hash value

cardano_node_tests.tests.test_configuration module

Tests for node configuration.

class cardano_node_tests.tests.test_configuration.TestBasic[source]

Bases: object

Basic tests for node configuration.

pytestmark = [Mark(name='noop', args=(), kwargs={}), Mark(name='noop', args=(), kwargs={}), Mark(name='skipif', args=(False,), kwargs={'reason': 'meant to run with default era or higher, where cluster era == Tx era'})][source]

test_epoch_length(cluster_epoch_length: ClusterLib)[source]

Test the epochLength configuration.

Test that custom epochLength parameter is correctly applied and epoch boundaries are respected.

Modify genesis spec to set epochLength to 1500 slots
Start cluster with modified configuration
Check that cluster slot length is 0.2 seconds (default)
Check that cluster epoch length is 1500 slots
Wait until end of current epoch
Check that epoch number increments correctly at epoch boundary
Verify slot count resets to 0 at epoch boundary

test_slot_length(cluster_slot_length: ClusterLib)[source]

Test the slotLength configuration.

Test that custom slotLength parameter is correctly applied and epoch boundaries are respected.

Modify genesis spec to set slotLength to 0.3 seconds
Start cluster with modified configuration
Check that cluster slot length is 0.3 seconds
Check that cluster epoch length is 1000 slots (default)
Wait until end of current epoch
Check that epoch number increments correctly at epoch boundary
Verify slot count resets to 0 at epoch boundary

cardano_node_tests.tests.test_configuration.check_epoch_length(cluster_obj: ClusterLib) → None[source]

cardano_node_tests.tests.test_configuration.cluster_epoch_length(cluster_manager: ClusterManager, epoch_length_start_cluster: Path) → ClusterLib[source]

cardano_node_tests.tests.test_configuration.cluster_slot_length(cluster_manager: ClusterManager, slot_length_start_cluster: Path) → ClusterLib[source]

cardano_node_tests.tests.test_configuration.epoch_length_start_cluster() → Path[source]: Update epochLength to 1200.

cardano_node_tests.tests.test_configuration.slot_length_start_cluster() → Path[source]: Update slotLength to 0.3.

cardano_node_tests.tests.test_dbsync module

Tests for db-sync.

class cardano_node_tests.tests.test_dbsync.TestDBSync[source]

Bases: object

General db-sync tests.

DBSYNC_TABLES: Final[set[str]] = {'ada_pots', 'block', 'collateral_tx_in', 'collateral_tx_out', 'committee', 'committee_de_registration', 'committee_hash', 'committee_member', 'committee_registration', 'constitution', 'cost_model', 'datum', 'delegation', 'delegation_vote', 'delisted_pool', 'drep_distr', 'drep_hash', 'drep_registration', 'epoch', 'epoch_param', 'epoch_stake', 'epoch_stake_progress', 'epoch_state', 'epoch_sync_time', 'extra_key_witness', 'extra_migrations', 'gov_action_proposal', 'ma_tx_mint', 'ma_tx_out', 'meta', 'multi_asset', 'off_chain_pool_data', 'off_chain_pool_fetch_error', 'off_chain_vote_author', 'off_chain_vote_data', 'off_chain_vote_drep_data', 'off_chain_vote_external_update', 'off_chain_vote_fetch_error', 'off_chain_vote_gov_action_data', 'off_chain_vote_reference', 'param_proposal', 'pool_hash', 'pool_metadata_ref', 'pool_owner', 'pool_relay', 'pool_retire', 'pool_update', 'pot_transfer', 'redeemer', 'redeemer_data', 'reference_tx_in', 'reserve', 'reserved_pool_ticker', 'reward', 'reward_rest', 'schema_version', 'script', 'slot_leader', 'stake_address', 'stake_deregistration', 'stake_registration', 'treasury', 'treasury_withdrawal', 'tx', 'tx_in', 'tx_metadata', 'tx_out', 'voting_anchor', 'voting_procedure', 'withdrawal'}[source]

test_blocks(cluster: ClusterLib)[source]

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

Test db-sync block table data integrity by validating field values and relationships across 50 epochs of block records.

Query current tip, block number, and epoch from cluster
Query block records from db-sync for last 50 epochs (epoch_from to current)
For each consecutive block pair, validate:
- id increases monotonically (rec.id > prev_rec.id)
- epoch_no is non-negative and increases monotonically
- epoch_no changes by at most 1 between consecutive blocks
- slot_no is non-negative and increases monotonically
- epoch_slot_no is non-negative and increases within same epoch
- block_no increments by exactly 1 between consecutive blocks
- previous_id points to previous record’s id
Check that last block_no in db-sync is within 1 of node’s block_no
Verify protocol version matches expected values

test_cost_model(cluster: ClusterLib)[source]

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

Test db-sync Plutus cost model records match protocol parameters. Runs only in eras > Alonzo where Plutus cost models are active.

Query current epoch number
Query cost models from db-sync for current epoch
If cost models not available, wait for next epoch and retry
Query protocol parameters from node for comparison
Extract PlutusV1, PlutusV2, PlutusV3 cost models from protocol params
Compare db-sync cost model records with protocol parameter cost models
Verify cost model values match between db-sync and node

test_epoch(cluster: ClusterLib)[source]

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

Test db-sync epoch table data consistency by comparing aggregated counts with block table data. Skip if running in epoch 0.

Query current epoch from cluster
Use previous epoch for validation (current - 1) if not in epoch 0
Skip test if in epoch 0
Query all blocks for the target epoch from block table
Count blocks and transactions from block table records
Query epoch summary from epoch table for same epoch
Count blocks and transactions from epoch table records
Verify block count matches between block table and epoch table
Verify transaction count matches between block table and epoch table

test_reconnect_dbsync(cluster_singleton: ClusterLib, cluster_manager: ClusterManager, worker_id: str)[source]

Check that db-sync reconnects to the node after the node is restarted.

Restart all nodes of the running cluster
Submit a transaction
Check that the transaction is present on dbsync

test_table_names(cluster: ClusterLib)[source]

Check that all the expected tables are present in db-sync.

Test db-sync database schema by verifying all expected tables exist.

Query db-sync database for list of all table names
Check that DBSYNC_TABLES set is a subset of actual table names
Verify all expected core tables are present (block, tx, tx_out, etc.)

class cardano_node_tests.tests.test_dbsync.TestDBSyncSnapshot[source]

Bases: object

Tests for db-sync snapshot availability and freshness.

test_latest_snapshot_freshness(cluster_manager: ClusterManager)[source]

Check that the latest db-sync snapshot is not older than 5 days.

This test uses the S3 REST API to query the Cardano mainnet snapshot repository and verifies that the most recent snapshot is fresh.

Initialize DBSyncSnapshotService to access S3 snapshot repository
Query S3 API to find latest db-sync version available
Get latest snapshot file for that version from S3
Log snapshot details (name, last modified date, size in GB)
Calculate current UTC time and 5-days-ago threshold
Verify snapshot last_modified date is within 5 days
Fail if snapshot is older than 5 days with detailed age information

cardano_node_tests.tests.test_dbsync_config module

Tests for basic DB-Sync configuration options.

class cardano_node_tests.tests.test_dbsync_config.ColumnCondition(*values)[source]

Bases: StrEnum

Enum for column-level db-sync condition checks.

IS_NULL = 'column_condition:IS NULL'[source]

ZERO = 'column_condition:=0'[source]

class cardano_node_tests.tests.test_dbsync_config.TableCondition(*values)[source]

Bases: StrEnum

Enum for table-level db-sync state conditions.

EMPTY = 'empty'[source]

EXISTS = 'exists'[source]

NOT_EMPTY = 'not_empty'[source]

NOT_EXISTS = 'not_exists'[source]

class cardano_node_tests.tests.test_dbsync_config.TestDBSyncConfig[source]

Bases: object

Tests for DB-Sync Config.

get_subtests() → Generator[Callable, None, None][source]

Get the DB-Sync Config scenarios.

The scenarios are executed as subtests in the test_dbsync_config test.

pytestmark = [Mark(name='order', args=(-1,), kwargs={})][source]

test_dbsync_config(cluster_singleton: ClusterLib, db_sync_manager: DBSyncManager, subtests: SubTests)[source]

Test DB-Sync configuration options using multiple subtests.

Verifies that different DB-Sync configuration settings correctly control table population and data insertion behavior. Each subtest modifies the configuration, restarts DB-Sync, and validates the expected database state.

Test tx_out option (enable/disable modes with various settings)
Verify address, tx_in, tx_out, and ma_tx_out tables respond to tx_out configuration
Test governance option (enable/disable)
Verify all governance-related tables populate when enabled and clear when disabled
Test tx_cbor option (enable/disable)
Verify tx_cbor table populates when enabled and clears when disabled
Test multi_asset option (enable/disable)
Verify multi_asset table populates when enabled and clears when disabled
Restore original DB-Sync configuration after all subtests complete

cardano_node_tests.tests.test_dbsync_config.check_dbsync_state(expected_state: dict[str | Table, TableCondition | ColumnCondition]) → None[source]

Check the state of db-sync tables and columns against expected conditions.

Parameters:

expected_state –

Dictionary specifying conditions to verify where:

Key format:
- ”table” for table-level checks
- ”table.column” for column-level checks
Value format:
- TableCondition enum values for table-level checks
- ColumnCondition enum values for column-level checks

Returns:

True if all conditions match, False otherwise

Return type:

bool

cardano_node_tests.tests.test_dbsync_config.db_sync_manager(cluster_singleton: ClusterLib) → Generator[DBSyncManager, None, None][source]

Provide db-sync manager on a singleton cluster.

Creates and returns a DBSyncManager instance with locked cluster resources to ensure exclusive access during testing.

Returns db-sync configuration to its original state after testing is finished.

cardano_node_tests.tests.test_delegation module

Tests for stake address delegation.

class cardano_node_tests.tests.test_delegation.TestDelegateAddr[source]

Bases: object

Tests for stake address delegation.

test_addr_delegation_deregistration(cluster_and_pool: tuple[ClusterLib, str], pool_users_cluster_and_pool: list[PoolUser], pool_users_disposable_cluster_and_pool: list[PoolUser], stake_cert: str, build_method: str)[source]

Submit delegation and deregistration certificates in single TX.

Create stake address registration cert
Create stake address deregistration cert
Register stake address
Create stake address delegation cert
Delegate and deregister stake address in single TX
Check that the balance for source address was correctly updated and that the key deposit was returned
Check that the stake address was NOT delegated
(optional) Check records in db-sync

test_delegate_multisig(cluster_manager: ClusterManager, cluster_and_pool_and_rewards: tuple[ClusterLib, str])[source]

Delegate and deregister multisig stake address.

Test delegation and deregistration of a stake address that uses a multisig script as stake credentials. All required signers must sign the transactions.

Create 5 stake key pairs
Create multisig script requiring all signatures (all type)
Create payment address using multisig script as stake credentials
Create script stake address using multisig script
Fund payment address
Generate stake address registration and delegation certificates
Sign and submit registration and delegation transaction with all required signatures
Check that the stake address is registered and delegated to pool
Wait 4 epochs for first reward
Attempt to deregister stake address with rewards - expect failure
Withdraw rewards to payment address and deregister stake address
Check that the key deposit was returned and rewards withdrawn
Check that the stake address is no longer delegated
(optional) Check records in db-sync

test_delegate_using_pool_id(cluster_manager: ClusterManager, cluster_and_pool: tuple[ClusterLib, str], build_method: str)[source]

Submit registration certificate and delegate to pool using pool id.

Uses parametrized build method (build or build-raw).

Create payment and stake address
Fund payment address
Generate stake address registration and delegation certificates using pool id
Submit registration and delegation certificates in single transaction
Check that the stake address was delegated to the expected pool
Check that the stake address is active and registered
(optional) Check records in db-sync

test_delegate_using_vkey(cluster_manager: ClusterManager, cluster_use_pool: tuple[ClusterLib, str], build_method: str)[source]

Submit registration certificate and delegate to pool using cold vkey.

Uses parametrized build method (build or build-raw). Delegates using pool cold vkey file instead of pool id.

Create payment and stake address
Fund payment address
Generate stake address registration and delegation certificates using pool cold vkey
Submit registration and delegation certificates in single transaction
Check that the stake address was delegated to the expected pool
Check that the stake address is active and registered
(optional) Check records in db-sync

test_deregister_delegated(cluster_manager: ClusterManager, cluster_and_pool_and_rewards: tuple[ClusterLib, str])[source]

Deregister a delegated stake address.

Create two payment addresses that share single stake address
Register and delegate the stake address to pool
Attempt to deregister the stake address - deregistration is expected to fail because there are rewards in the stake address
Withdraw rewards to payment address and deregister stake address
Check that the key deposit was returned and rewards withdrawn
Check that the stake address is no longer delegated
(optional) Check records in db-sync

test_multi_delegation(cluster_manager: ClusterManager, cluster_and_two_pools: tuple[ClusterLib, str, str])[source]

Delegate multiple stake addresses that share the same payment keys to multiple pools.

Test that multiple stake addresses using the same payment key can be delegated to different pools and all receive rewards independently.

Create 1 payment vkey/skey key pair
Create 4 stake vkey/skey key pairs
Create 4 payment addresses for each combination of payment_vkey/stake_vkey
Fund the payment addresses
Generate stake address registration certificates for all 4 stake addresses
Generate delegation certificates for the stake addresses (alternating between 2 pools)
Submit registration and delegation certificates in single transaction
Check that all stake addresses are delegated to their respective pools
Check that the balance for source address was correctly updated
Wait 4 epochs for first rewards
Check that all stake addresses received rewards from their respective pools

test_undelegate(cluster_manager: ClusterManager, cluster_and_pool_and_rewards: tuple[ClusterLib, str])[source]

Undelegate stake address.

Submit registration certificate and delegate to pool
Wait for first reward
Undelegate stake address:
- Withdraw rewards to payment address
- Deregister stake address
- Re-register stake address
Check that the key deposit was not returned
Check that rewards were withdrawn
Check that the stake address is still registered
Check that the stake address is no longer delegated
(optional) Check records in db-sync

class cardano_node_tests.tests.test_delegation.TestNegative[source]

Bases: object

Tests that are expected to fail.

test_delegate_addr_with_wrong_key(cluster_and_pool: tuple[ClusterLib, str], pool_users_cluster_and_pool: list[PoolUser], pool_users_disposable_cluster_and_pool: list[PoolUser])[source]

Try to delegate stake address using wrong payment skey.

Expect failure.

Create and register stake address
Generate stake address delegation certificate
Attempt to submit delegation transaction signed with wrong payment skey
Check that transaction submission fails with MissingVKeyWitnessesUTXOW error

test_delegate_deregistered_addr(cluster_and_pool: tuple[ClusterLib, str], pool_users_cluster_and_pool: list[PoolUser], pool_users_disposable_cluster_and_pool: list[PoolUser], build_method: str)[source]

Try to delegate deregistered stake address.

Expect failure.

Uses parametrized build method (build or build-raw).

Create and register stake address
Deregister the stake address
Generate stake address delegation certificate for deregistered address
Attempt to submit delegation transaction for deregistered stake address
Check that transaction fails with StakeDelegationImpossibleDELEG or StakeKeyNotRegisteredDELEG error

test_delegate_unknown_addr(cluster_and_pool: tuple[ClusterLib, str], pool_users_cluster_and_pool: list[PoolUser], pool_users_disposable_cluster_and_pool: list[PoolUser], build_method: str)[source]

Try to delegate unknown stake address.

Expect failure.

Uses parametrized build method (build or build-raw).

Create stake address but do not register it
Generate stake address delegation certificate for unregistered address
Attempt to submit delegation transaction for unknown stake address
Check that transaction fails with StakeDelegationImpossibleDELEG or StakeKeyNotRegisteredDELEG error

test_delegatee_not_registered(cluster: ClusterLib, pool_users: list[PoolUser], pool_users_disposable: list[PoolUser])[source]

Try to delegate stake address to unregistered pool.

Expect failure.

Create and register stake address
Generate pool cold keys and certificate but do not register the pool
Generate stake address delegation certificate using unregistered pool cold vkey
Attempt to submit delegation transaction to unregistered pool
Check that transaction fails with DelegateeNotRegisteredDELEG or DelegateeStakePoolNotRegisteredDELEG error

test_delegation_cert_with_wrong_key(cluster_and_pool: tuple[ClusterLib, str], pool_users_cluster_and_pool: list[PoolUser])[source]

Try to generate stake address delegation certificate using wrong stake vkey.

Expect failure.

Attempt to generate stake address delegation certificate using payment vkey instead of stake vkey
Check that certificate generation fails with expected error

cardano_node_tests.tests.test_delegation.cluster_and_pool(cluster_manager: ClusterManager) → tuple[ClusterLib, str][source]

cardano_node_tests.tests.test_delegation.cluster_and_pool_and_rewards(cluster_manager: ClusterManager) → tuple[ClusterLib, str][source]

cardano_node_tests.tests.test_delegation.cluster_and_two_pools(cluster_manager: ClusterManager) → tuple[ClusterLib, str, str][source]: Return instance of clusterlib.ClusterLib and two pools.

cardano_node_tests.tests.test_delegation.pool_users(cluster_manager: ClusterManager, cluster: ClusterLib) → list[PoolUser][source]: Create pool users.

cardano_node_tests.tests.test_delegation.pool_users_cluster_and_pool(cluster_manager: ClusterManager, cluster_and_pool: tuple[ClusterLib, str]) → list[PoolUser][source]: Create pool users using cluster_and_pool fixture.

Warning

The cached addresses can be used only for payments, not for delegation! The pool can be different every time the fixture is called.

cardano_node_tests.tests.test_delegation.pool_users_disposable(cluster: ClusterLib) → list[PoolUser][source]: Create function scoped pool users.

cardano_node_tests.tests.test_delegation.pool_users_disposable_cluster_and_pool(cluster_and_pool: tuple[ClusterLib, str]) → list[PoolUser][source]: Create function scoped pool users using cluster_and_pool fixture.

cardano_node_tests.tests.test_env_network_id module

Tests for CARDANO_NODE_NETWORK_ID environment variable.

class cardano_node_tests.tests.test_env_network_id.TestNegativeNetworkIdEnv[source]

Bases: object

Negative tests for CARDANO_NODE_NETWORK_ID.

test_neg_build_transfer_funds(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib, payment_addrs: list[AddressRecord], env_scenario: str, arg_scenario: str)[source]

Build transfer transaction with incorrect network configuration.

Test parametrized scenarios where CARDANO_NODE_NETWORK_ID env variable and/or –testnet-magic argument are missing or contain wrong values.

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

Create 2 payment addresses and fund first address
Set CARDANO_NODE_NETWORK_ID to wrong value or remove it (env_scenario parameter)
Set –testnet-magic to wrong value or remove it (arg_scenario parameter)
Attempt to build transaction sending 1.5 ADA from source to destination address
Check that transaction build fails with expected error message
Verify error is HandshakeError when network magic mismatch
Verify error is “Missing: (–mainnet | –testnet-magic NATURAL)” when both missing

Expect failure.

test_neg_query_pool_state(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib, env_scenario: str, arg_scenario: str)[source]

Query pool state with incorrect network configuration.

Test parametrized scenarios where CARDANO_NODE_NETWORK_ID env variable and/or –testnet-magic argument are missing or contain wrong values.

Set CARDANO_NODE_NETWORK_ID to wrong value or remove it (env_scenario parameter)
Set –testnet-magic to wrong value or remove it (arg_scenario parameter)
Execute cardano-cli query pool-state command for specific pool ID
Check that command fails with expected error message
Verify error is HandshakeError when network magic mismatch
Verify error is “Missing: (–mainnet | –testnet-magic NATURAL)” when both missing

Expect failure.

test_neg_query_protocol_params(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib, env_scenario: str, arg_scenario: str)[source]

Query protocol parameters with incorrect network configuration.

Test parametrized scenarios where CARDANO_NODE_NETWORK_ID env variable and/or –testnet-magic argument are missing or contain wrong values.

Set CARDANO_NODE_NETWORK_ID to wrong value or remove it (env_scenario parameter)
Set –testnet-magic to wrong value or remove it (arg_scenario parameter)
Execute cardano-cli query protocol-parameters command
Check that command fails with expected error message
Verify error is HandshakeError when network magic mismatch
Verify error is “Missing: (–mainnet | –testnet-magic NATURAL)” when both missing

Expect failure.

test_neg_query_protocol_state(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib, env_scenario: str, arg_scenario: str)[source]

Query protocol state with incorrect network configuration.

Test parametrized scenarios where CARDANO_NODE_NETWORK_ID env variable and/or –testnet-magic argument are missing or contain wrong values.

Set CARDANO_NODE_NETWORK_ID to wrong value or remove it (env_scenario parameter)
Set –testnet-magic to wrong value or remove it (arg_scenario parameter)
Execute cardano-cli query protocol-state command
Check that command fails with expected error message
Verify error is HandshakeError when network magic mismatch
Verify error is “Missing: (–mainnet | –testnet-magic NATURAL)” when both missing

Expect failure.

test_neg_query_stake_addr_info(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib, env_scenario: str, arg_scenario: str)[source]

Query stake address info with incorrect network configuration.

Test parametrized scenarios where CARDANO_NODE_NETWORK_ID env variable and/or –testnet-magic argument are missing or contain wrong values.

Set CARDANO_NODE_NETWORK_ID to wrong value or remove it (env_scenario parameter)
Set –testnet-magic to wrong value or remove it (arg_scenario parameter)
Execute cardano-cli query stake-address-info command for specific stake address
Check that command fails with expected error message
Verify error is HandshakeError when network magic mismatch
Verify error is “Missing: (–mainnet | –testnet-magic NATURAL)” when both missing

Expect failure.

test_neg_query_stake_distribution(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib, env_scenario: str, arg_scenario: str)[source]

Query stake distribution with incorrect network configuration.

Test parametrized scenarios where CARDANO_NODE_NETWORK_ID env variable and/or –testnet-magic argument are missing or contain wrong values.

Set CARDANO_NODE_NETWORK_ID to wrong value or remove it (env_scenario parameter)
Set –testnet-magic to wrong value or remove it (arg_scenario parameter)
Execute cardano-cli query stake-distribution command
Check that command fails with expected error message
Verify error is HandshakeError when network magic mismatch
Verify error is “Missing: (–mainnet | –testnet-magic NATURAL)” when both missing

Expect failure.

class cardano_node_tests.tests.test_env_network_id.TestNetworkIdEnv[source]

Bases: object

Tests for CARDANO_NODE_NETWORK_ID.

test_build_transfer_funds(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib, payment_addrs: list[AddressRecord])[source]

Send funds to payment address.

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

Send funds from 1 source address to 1 destination address
Check expected balances for both source and destination addresses

test_query_pool_state(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib)[source]

Query pool state using CARDANO_NODE_NETWORK_ID environment variable.

Test that cardano-cli can query pool state when network is specified via CARDANO_NODE_NETWORK_ID env variable instead of –testnet-magic argument.

Set CARDANO_NODE_NETWORK_ID environment variable to network magic
Remove –testnet-magic argument from cardano-cli commands
Execute cardano-cli query pool-state command for specific pool ID
Verify query succeeds without –testnet-magic argument

test_query_protocol_params(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib)[source]

Query protocol parameters using CARDANO_NODE_NETWORK_ID environment variable.

Test that cardano-cli can query protocol parameters when network is specified via CARDANO_NODE_NETWORK_ID env variable instead of –testnet-magic argument.

Set CARDANO_NODE_NETWORK_ID environment variable to network magic
Remove –testnet-magic argument from cardano-cli commands
Execute cardano-cli query protocol-parameters command
Check that parameters contain protocolVersion field
Verify query succeeds without –testnet-magic argument

test_query_protocol_state(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib)[source]

Query protocol state using CARDANO_NODE_NETWORK_ID environment variable.

Test that cardano-cli can query protocol state when network is specified via CARDANO_NODE_NETWORK_ID env variable instead of –testnet-magic argument.

Set CARDANO_NODE_NETWORK_ID environment variable to network magic
Remove –testnet-magic argument from cardano-cli commands
Execute cardano-cli query protocol-state command
Check that protocol state contains lastSlot field
Verify query succeeds without –testnet-magic argument

test_query_stake_addr_info(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib)[source]

Query stake address info using CARDANO_NODE_NETWORK_ID environment variable.

Test that cardano-cli can query stake address info when network is specified via CARDANO_NODE_NETWORK_ID env variable instead of –testnet-magic argument.

Set CARDANO_NODE_NETWORK_ID environment variable to network magic
Remove –testnet-magic argument from cardano-cli commands
Execute cardano-cli query stake-address-info command for specific stake address
Verify query succeeds without –testnet-magic argument

test_query_stake_distribution(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib)[source]

Query stake distribution using CARDANO_NODE_NETWORK_ID environment variable.

Test that cardano-cli can query stake distribution when network is specified via CARDANO_NODE_NETWORK_ID env variable instead of –testnet-magic argument.

Set CARDANO_NODE_NETWORK_ID environment variable to network magic
Remove –testnet-magic argument from cardano-cli commands
Execute cardano-cli query stake-distribution command
Check that distribution contains pool IDs starting with “pool”
Verify query succeeds without –testnet-magic argument

cardano_node_tests.tests.test_env_network_id.payment_addrs(skip_on_no_env: None, set_network_id_env: None, cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

cardano_node_tests.tests.test_env_network_id.set_network_id_env(cluster: ClusterLib) → Generator[None, None, None][source]: Set CARDANO_NODE_NETWORK_ID and prevent cardano-cli from using –testnet-magic.

cardano_node_tests.tests.test_env_network_id.skip_on_no_env(cluster: ClusterLib, set_network_id_env: None) → None[source]: Skip test if CARDANO_NODE_NETWORK_ID is not available.

cardano_node_tests.tests.test_kes module

Tests for KES period.

class cardano_node_tests.tests.test_kes.TestKES[source]

Bases: object

Basic tests for KES period.

test_expired_kes(cluster_kes: ClusterLib, cluster_manager: ClusterManager, worker_id: str)[source]

Test expired KES.

Start local cluster instance configured with short KES period and low number of key evolutions, so KES expires soon on all pools
Refresh opcert on 2 of the 3 pools, so KES doesn’t expire on those 2 pools and the pools keep forging blocks
Wait for KES expiration on the selected pool
Check that the pool with expired KES didn’t forge blocks in an epoch that followed after KES expiration
Check KES period info command with an operational certificate with an expired KES
Check KES period info command with operational certificates with a valid KES

test_negative_kes_period_arg(cluster: ClusterLib, cluster_manager: ClusterManager)[source]

Try to generate new operational certificate with a negative value for –kes-period.

Test that operational certificate generation fails when –kes-period is set to a negative value. Expect failure.

Get pool KES verification key, cold signing key, and counter file from pool2
Attempt to generate operational certificate with –kes-period set to -100
Check that generation fails with “KES_PERIOD must not be less than 0” error

test_no_kes_period_arg(cluster: ClusterLib, cluster_manager: ClusterManager)[source]

Try to generate new operational certificate without specifying the –kes-period.

Test that operational certificate generation fails when required –kes-period argument is not provided. Expect failure.

Get pool KES verification key, cold signing key, and counter file from pool2
Attempt to execute cardano-cli node issue-op-cert without –kes-period argument
Check that command fails with “Missing: –kes-period NATURAL” error
Verify operational certificate file was not created

test_opcert_invalid_kes_period(cluster_singleton: ClusterLib, cluster_manager: ClusterManager)[source]

Start a stake pool with an operational certificate created with invalid –kes-period.

Generate new operational certificate with –kes-period in the future
Restart the node with the new operational certificate
Check that the pool is not forging any blocks
If network era > Alonzo
- Generate new operational certificate with valid –kes-period, but counter value +2 from last used operational certificate
- Restart the node
- Check that the pool is not forging any blocks
Generate new operational certificate with valid –kes-period and restart the node
Check that the pool is forging blocks again

test_update_valid_opcert(cluster_singleton: ClusterLib, cluster_manager: ClusterManager)[source]

Update a valid operational certificate with another valid operational certificate.

Generate new operational certificate with valid –kes-period
Copy new operational certificate to the node
Stop the node so the corresponding pool is not forging new blocks
Check kes-period-info while the pool is not forging blocks
Start the node with the new operational certificate
Check that the pool is forging blocks again
Check that metrics reported by kes-period-info got updated once the pool started forging blocks again
Check kes-period-info with the old (replaced) operational certificate

cardano_node_tests.tests.test_kes.cluster_kes(cluster_manager: ClusterManager, short_kes_start_cluster: Path) → ClusterLib[source]

cardano_node_tests.tests.test_kes.short_kes_start_cluster() → Path[source]: Update slotsPerKESPeriod and maxKESEvolutions.

cardano_node_tests.tests.test_ledger_state module

Tests for ledger state.

class cardano_node_tests.tests.test_ledger_state.TestLedgerState[source]

Bases: object

Basic tests for ledger state.

test_stake_snapshot(cluster: ClusterLib)[source]

Test the stake-snapshot and ledger-state commands and ledger state values.

Test stake snapshot consistency by comparing ledger state data with stake-snapshot command output for all three stake snapshots (mark, set, go).

Wait for epoch interval (5 seconds into epoch) to ensure stable state
Get list of stake pool IDs from cluster
Skip if no pools available or too many pools (>200) on testnet
Query ledger state using query ledger-state command
Save ledger state to file for analysis
Extract stake snapshots (pstakeMark, pstakeSet, pstakeGo) from ledger state
For each snapshot, extract stake address hashes and delegations to pools
For each pool with delegations:
- Calculate total stake from ledger state for each snapshot (mark, set, go)
- Query stake snapshot using query stake-snapshot command
- Compare ledger state totals with stake-snapshot command output
- Verify stake amounts match between ledger state and stake-snapshot
Verify ledger state has expected top-level keys
Check that all delegated stake addresses are accounted for
Verify total active stake matches sum of individual pool stakes
Save all validation data to pickle file if errors found

cardano_node_tests.tests.test_metrics module

Tests for Prometheus and EKG metrics.

class cardano_node_tests.tests.test_metrics.TestEKG[source]

Bases: object

EKG metrics tests.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'meant to run with default era or higher, where cluster era == Tx era'})][source]

test_available_metrics(wait_epochs)[source]

Test that available EKG metrics matches the expected schema.

Test EKG (Event-driven Key-value) metrics endpoint to verify metrics conform to expected Pydantic model schema.

Wait for epoch >= 4 to ensure stable metrics
Get EKG port for pool1 from cluster configuration
Fetch metrics from EKG HTTP endpoint (GET / with Accept: application/json header)
Parse JSON response containing EKG metrics
Validate response against Pydantic Model schema
Check that all required fields are present and have correct types

class cardano_node_tests.tests.test_metrics.TestPrometheus[source]

Bases: object

Prometheus metrics tests.

EXPECTED_METRICS: Final[set[str]] = {'cardano_node_metrics_Forge_adopted_int', 'cardano_node_metrics_Forge_forge_about_to_lead_int', 'cardano_node_metrics_Forge_forged_int', 'cardano_node_metrics_Forge_node_is_leader_int', 'cardano_node_metrics_Forge_node_not_leader_int', 'cardano_node_metrics_Mem_resident_int', 'cardano_node_metrics_RTS_gcLiveBytes_int', 'cardano_node_metrics_RTS_gcMajorNum_int', 'cardano_node_metrics_RTS_gcMinorNum_int', 'cardano_node_metrics_RTS_gcticks_int', 'cardano_node_metrics_RTS_mutticks_int', 'cardano_node_metrics_Stat_cputicks_int', 'cardano_node_metrics_Stat_threads_int', 'cardano_node_metrics_blockNum_int', 'cardano_node_metrics_blocksForgedNum_int', 'cardano_node_metrics_currentKESPeriod_int', 'cardano_node_metrics_delegMapSize_int', 'cardano_node_metrics_density_real', 'cardano_node_metrics_epoch_int', 'cardano_node_metrics_mempoolBytes_int', 'cardano_node_metrics_myBlocksUncoupled_int', 'cardano_node_metrics_nodeIsLeaderNum_int', 'cardano_node_metrics_nodeStartTime_int', 'cardano_node_metrics_operationalCertificateExpiryKESPeriod_int', 'cardano_node_metrics_operationalCertificateStartKESPeriod_int', 'cardano_node_metrics_remainingKESPeriods_int', 'cardano_node_metrics_served_header_counter_int', 'cardano_node_metrics_slotInEpoch_int', 'cardano_node_metrics_slotNum_int', 'cardano_node_metrics_txsInMempool_int', 'cardano_node_metrics_txsProcessedNum_int', 'cardano_node_metrics_utxoSize_int', 'ekg_server_timestamp_ms', 'rts_gc_bytes_allocated', 'rts_gc_bytes_copied', 'rts_gc_cpu_ms', 'rts_gc_cumulative_bytes_used', 'rts_gc_current_bytes_slop', 'rts_gc_current_bytes_used', 'rts_gc_gc_cpu_ms', 'rts_gc_gc_wall_ms', 'rts_gc_init_cpu_ms', 'rts_gc_init_wall_ms', 'rts_gc_max_bytes_slop', 'rts_gc_max_bytes_used', 'rts_gc_mutator_cpu_ms', 'rts_gc_mutator_wall_ms', 'rts_gc_num_bytes_usage_samples', 'rts_gc_num_gcs', 'rts_gc_par_avg_bytes_copied', 'rts_gc_par_max_bytes_copied', 'rts_gc_par_tot_bytes_copied', 'rts_gc_peak_megabytes_allocated', 'rts_gc_wall_ms'}[source]

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'meant to run with default era or higher, where cluster era == Tx era'})][source]

test_available_metrics(wait_epochs)[source]

Test that list of available Prometheus metrics equals list of expected metrics.

Test Prometheus metrics endpoint to verify all expected metrics are exposed and no unexpected metrics are present.

Wait for epoch >= 4 to ensure stable metrics
Get Prometheus port for pool1 from cluster configuration
Fetch metrics from Prometheus HTTP endpoint (GET /metrics)
Parse metrics response text into individual metric names
Extract metric keys from response
Compare available metrics set with EXPECTED_METRICS set
Verify exact match (no missing or extra metrics)

cardano_node_tests.tests.test_metrics.get_ekg_metrics(port: int) → Response[source]

cardano_node_tests.tests.test_metrics.get_prometheus_metrics(port: int) → Response[source]

cardano_node_tests.tests.test_metrics.wait_epochs(cluster: ClusterLib)[source]: Make sure we are not checking metrics in epoch < 4.

cardano_node_tests.tests.test_mir_certs module

Tests for MIR certificates.

class cardano_node_tests.tests.test_mir_certs.TestMIRCerts[source]

Bases: object

Tests for MIR certificates.

test_build_pay_stake_addr_from(skip_on_hf_shortcut: None, cluster_manager: ClusterManager, cluster_pots: ClusterLib, registered_users: list[PoolUser], fund_src: str)[source]

Send funds from the reserves or treasury pot to stake address.

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

Generate an MIR certificate
Submit a TX with the MIR certificate
Check that the expected amount was added to the stake address reward account
(optional) Check transaction in db-sync

test_build_transfer_to_reserves(skip_on_hf_shortcut: None, cluster_manager: ClusterManager, cluster_pots: ClusterLib, pool_users: list[PoolUser])[source]

Send funds from the treasury pot to the reserves pot.

Test transferring funds between protocol pots using MIR certificate. Uses cardano-cli transaction build command for building the transactions.

Generate MIR certificate for transferring 1000000000000 lovelace to reserves
Build transaction using transaction build with MIR certificate
Sign transaction with genesis delegate keys
Submit transaction at the beginning of an epoch
Check that source address balance is correct after fee deduction
Check transaction view command output

test_build_transfer_to_treasury(skip_on_hf_shortcut: None, cluster_manager: ClusterManager, cluster_pots: ClusterLib, pool_users: list[PoolUser])[source]

Send funds from the reserves pot to the treasury pot.

Test transferring funds between protocol pots using MIR certificate. Uses cardano-cli transaction build command for building the transactions.

Generate MIR certificate for transferring 10000000000000 lovelace to treasury
Build transaction using transaction build with MIR certificate
Sign transaction with genesis delegate keys
Submit transaction at the beginning of an epoch
Check that source address balance is correct after fee deduction
Check transaction view command output

test_pay_multi_stake_addrs(skip_on_hf_shortcut: None, cluster_manager: ClusterManager, cluster_pots: ClusterLib, registered_users: list[PoolUser])[source]

Send funds from the reserves and treasury pots to multiple stake addresses in single TX.

Generate an MIR certificates for transferring from treasury for each stake address
Generate an MIR certificates for transferring from reserves for each stake address
Submit a TX with all the MIR certificates generated in previous steps
Check that the expected amount was added to all stake address reward accounts
(optional) Check transaction in db-sync

test_pay_stake_addr_from(skip_on_hf_shortcut: None, cluster_manager: ClusterManager, cluster_pots: ClusterLib, registered_users: list[PoolUser], fund_src: str)[source]

Send funds from the reserves or treasury pot to stake address.

Generate an MIR certificate
Submit a TX with the MIR certificate
Check that the expected amount was added to the stake address reward account
(optional) Check transaction in db-sync

test_pay_stake_addr_from_both(skip_on_hf_shortcut: None, cluster_manager: ClusterManager, cluster_pots: ClusterLib, registered_users: list[PoolUser])[source]

Send funds from the reserves and treasury pots to stake address.

Generate an MIR certificate for transferring from treasury
Generate an MIR certificate for transferring from reserves
Submit a TX with the treasury MIR certificate
In the same epoch as the previous TX, submit a TX with the reserves MIR certificate
Check that the expected amount was added to the stake address reward account
(optional) Check transactions in db-sync

test_pay_unregistered_stake_addr_from(skip_on_hf_shortcut: None, cluster_manager: ClusterManager, cluster_pots: ClusterLib, pool_users: list[PoolUser], fund_src: str, addr_history: str)[source]

Send funds from the reserves or treasury pot to unregistered stake address.

Generate an MIR certificate
If a stake address should be known on blockchain:
- Register the stake address
- If transferring funds from treasury, deregister the stake address BEFORE submitting the TX
Submit a TX with the MIR certificate
If a stake address should be known on blockchain and if transferring funds from reserves, deregister the stake address AFTER submitting the TX
Check that the amount was NOT added to the stake address reward account
(optional) Check transaction in db-sync

test_transfer_to_reserves(skip_on_hf_shortcut: None, cluster_manager: ClusterManager, cluster_pots: ClusterLib, pool_users: list[PoolUser])[source]

Send funds from the treasury pot to the reserves pot.

Test transferring funds between protocol pots using MIR certificate. Uses cardano-cli transaction build-raw. Expected to fail when Era < Alonzo.

Generate MIR certificate for transferring 1000000000000 lovelace to reserves
Submit transaction with MIR certificate signed by genesis delegate keys
Submit transaction at the beginning of an epoch
Check that source address balance is correct after fee deduction
Check transaction view command output

test_transfer_to_treasury(skip_on_hf_shortcut: None, cluster_manager: ClusterManager, cluster_pots: ClusterLib, pool_users: list[PoolUser])[source]

Send funds from the reserves pot to the treasury pot.

Test transferring funds between protocol pots using MIR certificate. Uses cardano-cli transaction build-raw. Expected to fail when Era < Alonzo.

Generate MIR certificate for transferring 10000000000000 lovelace to treasury
Submit transaction with MIR certificate signed by genesis delegate keys
Submit transaction at the beginning of an epoch
Check that source address balance is correct after fee deduction
Check transaction view command output

class cardano_node_tests.tests.test_mir_certs.TestNegativeMIRCerts[source]

Bases: object

Negative tests for MIR certificates.

test_exceed_pay_stake_addr_from(cluster_pots: ClusterLib, registered_users: list[PoolUser], fund_src: str)[source]

Try to send more funds than available from the reserves or treasury pot to stake address.

Expect failure.

Generate an MIR certificate
Submit a TX with the MIR certificate
Check that submitting the transaction fails with an expected error

cardano_node_tests.tests.test_mir_certs.cluster_pots(cluster_manager: ClusterManager) → ClusterLib[source]

cardano_node_tests.tests.test_mir_certs.pool_users(cluster_manager: ClusterManager, cluster_pots: ClusterLib) → list[PoolUser][source]: Create pool user.

cardano_node_tests.tests.test_mir_certs.registered_users(cluster_manager: ClusterManager, cluster_pots: ClusterLib, pool_users: list[PoolUser]) → list[PoolUser][source]: Register pool user’s stake address.

cardano_node_tests.tests.test_mir_certs.skip_on_hf_shortcut(cluster_pots: ClusterLib) → None[source]: Skip test if HF shortcut is used.

cardano_node_tests.tests.test_mnemonic module

Tests for deriving keys from a mnemonic sentence.

class cardano_node_tests.tests.test_mnemonic.BadMnemonicCase(name: str, content_bytes: bytes | None, kind: Literal['file', 'empty_file', 'no_read_perm', 'directory', 'broken_symlink', 'symlink_loop'])[source]

Bases: object

A single non-compliant mnemonic-file test case.

content_bytes: bytes | None[source]

kind: Literal['file', 'empty_file', 'no_read_perm', 'directory', 'broken_symlink', 'symlink_loop'][source]

name: str[source]

class cardano_node_tests.tests.test_mnemonic.TestMnemonic[source]

Bases: object

Tests for mnemonic sentence.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'meant to run with default era or higher, where cluster era == Tx era'})][source]

test_derive_account_key_number_property(cluster: ClusterLib, key_type: KeyType) → None[source]

Test that derive-from-mnemonic accepts any valid account_number in [0, 2^31-1].

Test key derivation with boundary values for account_number to ensure full range of valid account numbers is supported. For payment/stake keys, pass the same value as key_number, otherwise omit key_number.

Use golden mnemonic file (24 words)
Derive key using parametrized key type
Use parametrized account_number (property-based test covering 0 to 2^31-1)
For payment/stake keys, set key_number to same value as account_number
For other key types (DRep, CC), omit key_number parameter
Verify derived key file exists

test_gen_and_deriv(cluster: ClusterLib, size: Literal[12, 15, 18, 21, 24], out: str, key_type: KeyType, out_format: OutputFormat, path_num: int)[source]

Test generating mnemonic and deriving keys from mnemonic.

Test mnemonic generation and key derivation with various parameters using generate-mnemonic and derive-from-mnemonic commands.

Generate mnemonic of parametrized size (12, 15, 18, 21, or 24 words)
Output mnemonic to file or stdout based on parametrization
Check that generated mnemonic has expected word count
Derive key from mnemonic using parametrized key type (payment, stake, DRep, etc.)
Use parametrized output format (bech32, hex, text-envelope)
Use parametrized account number and key number (path_num: 0 or 2^31-1)
Verify derived key file exists

test_golden_deriv(cluster: ClusterLib, size: Literal[12, 15, 18, 21, 24], key_type: KeyType, out_format: OutputFormat, path_num: int)[source]

Test deriving keys from mnemonic using golden files.

Test key derivation against known golden mnemonic files to ensure consistent key generation across different parameters.

Use golden mnemonic file of parametrized size (12, 15, 18, 21, or 24 words)
Derive key using parametrized key type (payment, stake, DRep, CC cold, CC hot)
Use parametrized output format (bech32, hex, text-envelope)
Use parametrized account number (path_num: 0 or 2^31-1)
Verify derived key file exists
Compare derived key against expected golden output file
Check that derived key matches expected value from golden file

class cardano_node_tests.tests.test_mnemonic.TestNegativeMnemonic[source]

Bases: object

Tests with invalid arguments.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'meant to run with default era or higher, where cluster era == Tx era'})][source]

test_gen_invalid_size(cluster: ClusterLib) → None[source]

Test generating a mnemonic with an invalid size.

Test that mnemonic generation fails when size is not one of the allowed values (12, 15, 18, 21, 24). Uses Hypothesis property-based testing. Expect failure.

Attempt to generate mnemonic with invalid size (any integer except 12,15,18,21,24)
Check that generation fails with “Invalid mnemonic size” error

test_reject_invalid_account_or_key_number(cluster: ClusterLib, key_type: KeyType) → None[source]

Property: derive_from_mnemonic rejects out-of-range or non-int account/key numbers.

Test that key derivation fails when account_number or key_number is invalid (out of range or wrong type). Uses Hypothesis property-based testing. Expect failure.

Use golden mnemonic file (24 words)
Attempt to derive key using invalid account_number or key_number value
Invalid values: negative integers, integers >= 2^31, floats, strings, objects
Check that derivation fails with appropriate error (CLIError or ValueError/TypeError)

test_rejects_noncompliant_mnemonics(cluster: ClusterLib, tmp_case_path: Path) → None[source]

Test that the CLI wrapper rejects malformed mnemonic files.

Test key derivation failure with various invalid mnemonic file formats using Hypothesis property-based testing. Expect failure.

Generate malformed mnemonic file (invalid cases from Hypothesis strategy)
Attempt to derive key from malformed mnemonic file
Check that derivation fails with error about reading mnemonic file or converting mnemonic into a key

tmp_case_path() → Generator[Path, None, None][source]

cardano_node_tests.tests.test_mnemonic.broken_symlink_case() → SearchStrategy[BadMnemonicCase][source]

cardano_node_tests.tests.test_mnemonic.directory_case() → SearchStrategy[BadMnemonicCase][source]

cardano_node_tests.tests.test_mnemonic.empty_file_case() → SearchStrategy[BadMnemonicCase][source]

cardano_node_tests.tests.test_mnemonic.encoding_garbage_files() → SearchStrategy[BadMnemonicCase][source]: Binary junk, odd encodings, NULs.

cardano_node_tests.tests.test_mnemonic.join_with_weirdness(*, tokens: Sequence[str]) → SearchStrategy[str][source]: Join tokens with odd/mixed separators and optional leading/trailing clutter.

cardano_node_tests.tests.test_mnemonic.maybe_reencode(text: str) → SearchStrategy[bytes][source]: Return bytes in various encodings and with potential binary junk.

cardano_node_tests.tests.test_mnemonic.no_read_perm_case() → SearchStrategy[BadMnemonicCase][source]

cardano_node_tests.tests.test_mnemonic.separator_mess_files() → SearchStrategy[BadMnemonicCase][source]: Only separators/newlines/tabs/comma issues (often two+ separators between words).

cardano_node_tests.tests.test_mnemonic.symlink_loop_case() → SearchStrategy[BadMnemonicCase][source]

cardano_node_tests.tests.test_mnemonic.unknown_word_files() → SearchStrategy[BadMnemonicCase][source]: Produce allowed word count, but at least one token is not a valid BIP39 word.

cardano_node_tests.tests.test_mnemonic.word_sequence(*, count: int, ensure_invalid: bool) → SearchStrategy[list[str]][source]

Build a list of tokens of a specific count.

If ensure_invalid=True, inject at least one invalid token.

cardano_node_tests.tests.test_mnemonic.wrong_count_files() → SearchStrategy[BadMnemonicCase][source]: Wrong number of words (not in {12, 15, 18, 21, 24}).

cardano_node_tests.tests.test_native_tokens module

Tests for native tokens.

Minting
Burning
Locking
Transactions

class cardano_node_tests.tests.test_native_tokens.TestCLITxOutSyntax[source]

Bases: object

Tests of syntax for specifying muti-asset values and txouts.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'meant to run with default era or higher, where cluster era == Tx era'})][source]

test_multiasset_txouts_syntax(cluster: ClusterLib, issuers_addrs: list[AddressRecord])[source]

Test syntax for specifying multi-asset values and txouts via CLI.

Test it by minting one token and burning the same token in single transaction.

Create a script
Specify amount to mint and amount to burn in the same transaction
Assemble CLI arguments for transaction build and test syntax for multi-asset values and txouts
Build Tx body using the assembled CLI arguments, sign and submit the Tx
Check that the expected amount was minted (to_mint_amount - to_burn_amount)
Check fees in Lovelace
(optional) Check transactions in db-sync

class cardano_node_tests.tests.test_native_tokens.TestMinting[source]

Bases: object

Tests for minting and burning tokens.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'native tokens are available only in Mary+ eras'})][source]

test_bundle_minting_and_burning_sign(cluster_manager: ClusterManager, cluster: ClusterLib, worker_id: str, issuers_addrs: list[AddressRecord], simple_script_policyid: tuple[Path, str], tokens_db: tuple[int, int, int], build_method: str, submit_method: str)[source]

Test minting and burning multiple different tokens that are in single bundle.

Sign the TX using skeys.

Mint several tokens using a single script
Burn the minted tokens
Check fees in Lovelace
(optional) Check transactions in db-sync

test_bundle_minting_and_burning_witnesses(cluster_manager: ClusterManager, cluster: ClusterLib, worker_id: str, issuers_addrs: list[AddressRecord], multisig_script_policyid: tuple[Path, str], tokens_db: tuple[int, int, int], build_method: str, submit_method: str)[source]

Test minting and burning multiple different tokens that are in single bundle.

Sign the TX using witnesses.

Mint several tokens using a single script
Burn the minted tokens
Check fees in Lovelace
(optional) Check transactions in db-sync

test_minting_and_burning_sign(cluster: ClusterLib, issuers_addrs: list[AddressRecord], aname_type: str, build_method: str, submit_method: str)[source]

Test minting and burning of tokens, sign the transaction using skeys.

Mint 2 tokens - one identified by policyid + asset name and one identified by just policyid
Burn the minted tokens
Check fees in Lovelace
(optional) Check transactions in db-sync

test_minting_and_burning_witnesses(cluster: ClusterLib, issuers_addrs: list[AddressRecord], aname_type: str, build_method: str, submit_method: str)[source]

Test minting and burning of tokens, sign the transaction using witnesses.

Mint 2 tokens - one identified by policyid + asset name and one identified by just policyid
Burn the minted tokens
Check fees in Lovelace
Check output of the transaction view command
(optional) Check transactions in db-sync

test_minting_and_partial_burning(cluster: ClusterLib, issuers_addrs: list[AddressRecord], build_method: str, submit_method: str)[source]

Test minting and partial burning of tokens.

Mint a token
Burn part of the minted token, check the expected amount
Check fees in Lovelace
(optional) Check transactions in db-sync

test_minting_burning_diff_tokens_single_tx(cluster: ClusterLib, issuers_addrs: list[AddressRecord], build_method: str, submit_method: str)[source]

Test minting one token and burning other token in single transaction.

Sign transactions using skeys.

Create a script
1st TX - mint first token
2nd TX - mint second token, burn first token
3rd TX - burn second token
Check fees in Lovelace
(optional) Check transactions in db-sync

test_minting_burning_same_token_single_tx(cluster: ClusterLib, issuers_addrs: list[AddressRecord], build_method: str, submit_method: str)[source]

Test minting one token and burning the same token in single transaction.

Sign transactions using skeys.

Create a script
Specify amount to mint and amount to burn in the same transaction
Check that the expected amount was minted (to_mint_amount - to_burn_amount)
Check fees in Lovelace
(optional) Check transactions in db-sync

test_minting_multiple_scripts(cluster: ClusterLib, issuers_addrs: list[AddressRecord], build_method: str, submit_method: str)[source]

Test minting of tokens using several different scripts in single transaction.

Create tokens issuers
Create a script for each issuer
Mint 2 tokens with each script - one identified by policyid + asset name and one identified by just policyid. The minting is done in single transaction, the transaction is signed using skeys.
Check that the tokens were minted
Burn the minted tokens
Check that the tokens were burnt
Check fees in Lovelace
(optional) Check transactions in db-sync

test_minting_unicode_asset_name(cluster: ClusterLib, issuers_addrs: list[AddressRecord], build_method: str, submit_method: str)[source]

Test minting and burning of token with unicode non-ascii chars in its asset name.

Tests https://github.com/IntersectMBO/cardano-node/issues/2337

Mint a token that has non-ascii characters in its asset name
Burn the minted token
Check fees in Lovelace
(optional) Check transactions in db-sync

class cardano_node_tests.tests.test_native_tokens.TestNegative[source]

Bases: object

Negative tests for minting tokens.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'native tokens are available only in Mary+ eras'})][source]

test_long_name(cluster: ClusterLib, issuers_addrs: list[AddressRecord], simple_script_policyid: tuple[Path, str]) → None[source]

Try to create token with asset name that is longer than allowed.

The name can also contain characters that are not allowed. Expect failure.

test_minting_amount_above_the_allowed(cluster: ClusterLib, issuers_addrs: list[AddressRecord], simple_script_policyid: tuple[Path, str]) → None[source]

Test minting a token amount above the maximum allowed.

Test that token minting fails when the amount exceeds the maximum allowed value for native tokens. Uses Hypothesis property-based testing. Expect failure.

Create native token with amount above MAX_TOKEN_AMOUNT (generated by Hypothesis)
Get simple script and policy ID from fixture
Attempt to mint token with excessive amount
Check that minting fails with “the number exceeds the max bound” error

class cardano_node_tests.tests.test_native_tokens.TestPolicies[source]

Bases: object

Tests for minting and burning tokens using minting policies.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'native tokens are available only in Mary+ eras'})][source]

test_policy_after_future(cluster: ClusterLib, issuers_addrs: list[AddressRecord])[source]

Test minting tokens with time-locking policy requiring future slot.

Test that tokens cannot be minted when the “after” slot constraint is in the future and transaction validity intervals don’t satisfy or violate the policy. Expect failure.

Create multisig script with “after slot” constraint set to future slot (current + 10000)
Generate policy ID from script
Create 5 different native tokens under same policy
Attempt to mint with valid range (invalid-before at future slot, invalid-hereafter future slot + 100) - fails because slot is in the future (OutsideValidityIntervalUTxO)
Attempt to mint with invalid range (invalid-before 1, invalid-hereafter at future slot) - fails because “after” constraint not met (ScriptWitnessNotValidatingUTXOW)
Verify tokens were not minted

test_policy_after_past(cluster: ClusterLib, issuers_addrs: list[AddressRecord])[source]

Test minting tokens with time-locking policy requiring already-passed slot.

Test that tokens cannot be minted when the “after” slot constraint references a past slot and the transaction validity interval violates the constraint. Expect failure.

Create multisig script with “after slot” constraint set to past slot (current - 1)
Generate policy ID from script
Create 5 different native tokens under same policy
Attempt to mint with invalid-before 1 and invalid-hereafter at past slot
Verify minting fails with ScriptWitnessNotValidatingUTXOW error
Verify tokens were not minted

test_policy_before_future(cluster: ClusterLib, issuers_addrs: list[AddressRecord])[source]

Attempt to mint tokens when transaction validity violates “before” policy constraint.

Test that minting fails when the time-locking policy “before” constraint is violated because transaction invalid-hereafter is after the policy threshold.

Create multisig script with “before slot N” where N is far future (current + 10000)
Generate policy ID from script
Create 5 native tokens under this policy
Attempt to mint with invalid-hereafter after the threshold slot
Verify minting fails with ScriptWitnessNotValidatingUTXOW error (policy not satisfied)
Check that tokens were not minted

Expect failure.

test_policy_before_past(cluster: ClusterLib, issuers_addrs: list[AddressRecord])[source]

Attempt to mint tokens when “before” slot constraint is already in the past.

Test that minting fails when the time-locking policy “before” constraint cannot be satisfied because the threshold slot is in the past.

Create multisig script with “before slot N” where N is current slot - 1 (past)
Generate policy ID from script
Create 5 native tokens under this policy
Attempt to mint with invalid-hereafter matching the past slot
Verify minting fails with OutsideValidityIntervalUTxO error
Attempt to mint with invalid-hereafter after the past slot
Verify minting fails with ScriptWitnessNotValidatingUTXOW error
Check that tokens were not minted

Expect failure.

test_valid_policy_after(cluster: ClusterLib, issuers_addrs: list[AddressRecord], build_method: str)[source]

Mint and burn native tokens with time-locking policy requiring slot after threshold.

Test that tokens can be minted and burned when transaction validity interval satisfies the “after” slot constraint in the minting policy script.

Create multisig script with “after slot 100” time-locking constraint
Generate policy ID from script
Create 5 different native tokens under same policy
Mint tokens with invalid-before set to slot 100 (satisfies “after” constraint)
Check tokens were minted at expected address
Burn tokens with same time constraint
Check tokens were burned
Verify transaction fees are within expected range
(optional) Check transactions in db-sync

test_valid_policy_before(cluster: ClusterLib, issuers_addrs: list[AddressRecord], build_method: str)[source]

Mint and burn native tokens with time-locking policy requiring slot before threshold.

Test that tokens can be minted and burned when transaction validity interval satisfies the “before” slot constraint in the minting policy script.

Create multisig script with “before slot N” time-locking constraint (future slot)
Generate policy ID from script
Create 5 different native tokens under same policy
Mint tokens with invalid-hereafter before the threshold (satisfies “before” constraint)
Check tokens were minted at expected address
Burn tokens with same time constraint
Check tokens were burned
Verify transaction fees are within expected range
(optional) Check transactions in db-sync

class cardano_node_tests.tests.test_native_tokens.TestReferenceUTxO[source]

Bases: object

Tests for Simple Scripts V1 and V2 on reference UTxOs.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'runs only with Babbage+ TX'})][source]

test_script_reference_utxo(cluster: ClusterLib, issuers_addrs: list[AddressRecord], build_method: str, script_version: str)[source]

Test minting and burning a token using reference script.

Mint and burn token in the same transaction Sign transactions using skeys.

Create a Simple Script
Create a reference UTxO with the script
Specify amount to mint and amount to burn in the same transaction
Check that the expected amount was minted (to_mint_amount - to_burn_amount)
Check fees in Lovelace
(optional) Check transactions in db-sync

class cardano_node_tests.tests.test_native_tokens.TestTransfer[source]

Bases: object

Tests for transferring tokens.

NEW_TOKENS_NUM = 20000000[source]

new_token(cluster_manager: ClusterManager, cluster: ClusterLib, payment_addrs: list[AddressRecord]) → NativeTokenRec[source]

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'native tokens are available only in Mary+ eras'})][source]

test_transfer_invalid_token_amount(cluster: ClusterLib, payment_addrs: list[AddressRecord], new_token: NativeTokenRec, build_method: str) → None[source]

Test sending an invalid amount of tokens to payment address.

Test that transactions fail when attempting to send more tokens than available or negative token amounts, depending on the transaction build method. Expect failure.

Create transaction output with invalid token amount (negative or exceeds balance)
Attempt to build/send transaction using parametrized build method
Check that transaction build/submission fails with appropriate error
For build method: error about unbalanced assets or illegal value
For build-raw method: error ValueNotConservedUTxO on submission
For build-estimate method: error about unbalanced assets or illegal value

test_transfer_multiple_tokens(cluster: ClusterLib, payment_addrs: list[AddressRecord], new_token: NativeTokenRec, build_method: str)[source]

Test sending multiple different tokens to payment addresses.

Send multiple different tokens from 1 source address to 2 destination addresses
Check expected token balances for both source and destination addresses for each token
Check fees in Lovelace
(optional) Check transactions in db-sync

test_transfer_no_ada(cluster: ClusterLib, payment_addrs: list[AddressRecord], new_token: NativeTokenRec, build_method: str)[source]: Try to create an UTxO with just native tokens, no ADA. Expect failure.

test_transfer_tokens(cluster: ClusterLib, payment_addrs: list[AddressRecord], new_token: NativeTokenRec, amount: int, build_method: str)[source]

Test sending tokens to payment address.

Send tokens from 1 source address to 1 destination address
Check expected token balances for both source and destination addresses
Check fees in Lovelace
(optional) Check transactions in db-sync

cardano_node_tests.tests.test_native_tokens.issuers_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new issuers addresses.

cardano_node_tests.tests.test_native_tokens.multisig_script_policyid(cluster: ClusterLib, issuers_addrs: list[AddressRecord]) → tuple[Path, str][source]: Return multisig script and it’s PolicyId.

cardano_node_tests.tests.test_native_tokens.simple_script_policyid(cluster: ClusterLib, issuers_addrs: list[AddressRecord]) → tuple[Path, str][source]: Return script and its PolicyId.

cardano_node_tests.tests.test_node_upgrade module

Tests for node upgrade.

class cardano_node_tests.tests.test_node_upgrade.TestSetup[source]

Bases: object

Tests for setting up cardano network before and during upgrade testing.

Special tests that run outside of normal test run.

pool_user_singleton(cluster_manager: ClusterManager, cluster_singleton: ClusterLib) → PoolUser[source]: Create a pool user for singleton.

test_hardfork(cluster_manager: ClusterManager, cluster_singleton: ClusterLib, pool_user_singleton: PoolUser)[source]

Test hard fork.

Test protocol version hard fork using governance action after node upgrade. Runs only on step 3 of upgrade testing sequence.

Get current protocol version and calculate target version (current + 1)
Skip if already at LAST_KNOWN_PROTOCOL_VERSION
Check that ExperimentalHardForksEnabled is true in node config
Get default governance data (DReps, committee members, pools)
Wait for any delayed ratification to complete
Create hardfork governance action with target protocol version
Submit hardfork action transaction
Vote on hardfork action (Constitutional Committee, DReps, SPOs)
Wait for ratification
Wait for enactment epoch
Verify protocol version changed to target version

test_ignore_log_errors(cluster_singleton: ClusterLib, worker_id: str)[source]

Ignore selected errors in log right after node upgrade.

Test setup to ignore expected errors in node logs after upgrading from pre-UTxO-HD releases. UTxO-HD was added in node 10.4.1.

Check if upgraded node version is < 10.5.0
If upgrading from pre-UTxO-HD release (node < 10.5.0):
- Add log ignore rule for ChainDB warning about invalid snapshot
- Ignore “MetadataFileDoesNotExist” error in stdout logs
This prevents false test failures from expected upgrade-related log messages

test_update_cost_models(cluster_manager: ClusterManager, cluster_singleton: ClusterLib, pool_user_singleton: PoolUser)[source]

Test cost model update.

Test updating Plutus cost models after node upgrade. Runs only on step 2 of upgrade testing sequence.

Load cost model proposal from JSON file (PlutusV2 and PlutusV3 models)
Get default governance data (DReps, committee members, pools)
Submit cost model update governance action
Vote and ratify the cost model update
Wait for enactment
Verify updated cost models are active

class cardano_node_tests.tests.test_node_upgrade.TestUpgrade[source]

Bases: object

Tests for node upgrade testing.

test_prepare_tx(cluster_manager: ClusterManager, cluster: ClusterLib, payment_addrs_disposable: list[AddressRecord], build_method: str, for_step: int, file_type: str)[source]

Prepare transactions that will be submitted in next steps of upgrade testing.

For testing that transaction created by previous node version and/or in previous era can be submitted in next node version and/or next era. Runs on steps 1-2, creates transactions for steps 2-3.

Create simple transaction sending 2 ADA from source to destination address
Build transaction using parametrized method (build or build-raw)
For build method: use fee buffer of 1000000 lovelace
For build-raw method: calculate fee first, then build transaction
Sign transaction with source address signing key
Save either signed tx file or unsigned tx body file based on parametrization
Save to state directory for use in later upgrade testing steps
Verify saved file exists and has non-zero size

test_submit_tx(cluster_manager: ClusterManager, cluster: ClusterLib, build_method: str, for_step: int, from_step: int, file_type: str)[source]

Submit transaction that was created by previous node version and/or in previous era.

Test cross-version/cross-era transaction compatibility by submitting transactions created in earlier upgrade steps. Runs on steps 2-3.

Locate transaction file created in previous step (from state directory)
Load signed tx file or unsigned tx body file based on parametrization
If loading tx body: sign the transaction body with appropriate signing keys
Submit the transaction to the upgraded node
Wait for transaction to be included in a block
Verify transaction was successfully submitted and processed
Check expected balances after transaction submission
Demonstrates backward compatibility of transaction format across versions/eras

cardano_node_tests.tests.test_node_upgrade.payment_addr_locked(cluster_manager: ClusterManager, cluster_singleton: ClusterLib) → AddressRecord[source]: Create new payment addresses.

cardano_node_tests.tests.test_node_upgrade.payment_addrs_disposable(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new disposable payment addresses.

cardano_node_tests.tests.test_pool_saturation module

Tests of effect of pool saturation on rewards and blocks production.

class cardano_node_tests.tests.test_pool_saturation.PoolRecord(name: str, id: str, id_dec: str, reward_addr: cardano_clusterlib.structs.PoolUser, delegation_out: cardano_node_tests.tests.delegation.DelegationOut, user_rewards: list[cardano_node_tests.tests.test_pool_saturation.RewardRecord], owner_rewards: list[cardano_node_tests.tests.test_pool_saturation.RewardRecord], blocks_forged: dict[int, int], saturation_amounts: dict[int, int])[source]

Bases: object

blocks_forged: dict[int, int][source]

delegation_out: DelegationOut[source]

id: str[source]

id_dec: str[source]

name: str[source]

owner_rewards: list[RewardRecord][source]

reward_addr: PoolUser[source]

saturation_amounts: dict[int, int][source]

user_rewards: list[RewardRecord][source]

class cardano_node_tests.tests.test_pool_saturation.RewardRecord(epoch_no: int, reward_total: int, reward_per_epoch: int, stake_total: int)[source]

Bases: object

epoch_no: int[source]

reward_per_epoch: int[source]

reward_total: int[source]

stake_total: int[source]

class cardano_node_tests.tests.test_pool_saturation.TestPoolSaturation[source]

Bases: object

test_oversaturated(cluster_manager: ClusterManager, cluster_lock_pools: ClusterLib)[source]

Check diminished rewards when stake pool is oversaturated.

The stake pool continues to operate normally and those who delegate to that pool receive rewards, but the rewards are proportionally lower than those received from stake pool that is not oversaturated.

Register and delegate stake address in “init epoch”, for all available pools
In “init epoch” + 2, saturate all available pools (block distribution remains balanced among pools)
In “init epoch” + 4, oversaturate one pool
In “init epoch” + 6, for all available pools, withdraw rewards and transfer funds from delegated addresses so pools are no longer (over)saturated
While doing the steps above, collect rewards data for 10 epochs
Compare proportionality of rewards in epochs where pools were non-saturated, saturated and oversaturated

cardano_node_tests.tests.test_pool_saturation.cluster_lock_pools(cluster_manager: ClusterManager) → ClusterLib[source]

cardano_node_tests.tests.test_pools module

Tests for operations with stake pools.

Pool registration
Pool deregistration
Pool update
Pool metadata
Pool reregistration

class cardano_node_tests.tests.test_pools.TestNegative[source]

Bases: object

Stake pool tests that are expected to fail.

gen_pool_registration_cert_data(cluster: ClusterLib) → tuple[str, str, KeyPair, ColdKeyPair][source]

pool_data() → PoolData[source]

pool_users(cluster_manager: ClusterManager, cluster: ClusterLib) → list[PoolUser][source]: Create pool users.

test_pool_deregistration_not_registered(cluster: ClusterLib, pool_users: list[PoolUser], pool_data: PoolData, build_method: str)[source]

Try to deregister pool that is not registered.

Expect failure.

test_pool_registration_cert_wrong_cold(cluster: ClusterLib, pool_users: list[PoolUser], pool_data: PoolData)[source]

Try to generate pool registration certificate using wrong Cold vkey.

Expect failure.

test_pool_registration_cert_wrong_stake(cluster: ClusterLib, pool_users: list[PoolUser], pool_data: PoolData)[source]

Try to generate pool registration certificate using wrong stake vkey.

Expect failure.

test_pool_registration_cert_wrong_vrf(cluster: ClusterLib, pool_users: list[PoolUser], pool_data: PoolData)[source]

Try to generate pool registration certificate using wrong VRF key.

Expect failure.

test_pool_registration_missing_cold_skey(cluster: ClusterLib, pool_users: list[PoolUser], pool_data: PoolData)[source]

Try to register pool using transaction with missing Cold skey.

Expect failure.

test_pool_registration_missing_payment_skey(cluster: ClusterLib, pool_users: list[PoolUser], pool_data: PoolData)[source]

Try to register pool using transaction with missing payment skey.

Expect failure.

test_stake_pool_long_metadata_url(cluster: ClusterLib, pool_users: list[PoolUser], gen_pool_registration_cert_data: tuple[str, str, KeyPair, ColdKeyPair]) → None[source]

Try to create pool registration cert when the metadata-url is longer than allowed.

Expect failure. Property-based test.

test_stake_pool_metadata_long_description(cluster: ClusterLib) → None[source]

Try to create pool metadata hash when the description value is longer than allowed.

Expect failure. Property-based test.

test_stake_pool_metadata_long_homepage(cluster: ClusterLib) → None[source]

Try to create pool metadata hash when the homepage value is longer than allowed.

Expect failure. Property-based test.

test_stake_pool_metadata_long_name(cluster: ClusterLib) → None[source]

Try to create pool metadata hash when the name value is longer than allowed.

Expect failure. Property-based test.

test_stake_pool_metadata_long_ticker(cluster: ClusterLib) → None[source]

Try to create pool metadata hash when the ticker value is longer than allowed.

Expect failure. Property-based test.

test_stake_pool_metadata_no_description(cluster: ClusterLib)[source]

Try to create pool metadata hash when missing the description key.

Expect failure.

test_stake_pool_metadata_no_homepage(cluster: ClusterLib)[source]

Try to create pool metadata hash when missing the homepage key.

Expect failure.

test_stake_pool_metadata_no_name(cluster: ClusterLib)[source]

Try to create pool metadata hash when missing the name key.

Expect failure.

test_stake_pool_metadata_no_ticker(cluster: ClusterLib)[source]

Try to create pool metadata hash when missing the ticker key.

Expect failure.

class cardano_node_tests.tests.test_pools.TestPoolCost[source]

Bases: object

Tests for stake pool cost.

cluster_mincost(cluster_manager: ClusterManager, pool_cost_start_cluster: Path) → ClusterLib[source]

pool_owners_pbt(cluster_manager: ClusterManager, cluster_mincost: ClusterLib)[source]: Create pool owners for property-based test.

pytestmark = [Mark(name='noop', args=(), kwargs={}), Mark(name='noop', args=(), kwargs={})][source]

test_stake_pool_cost(cluster_manager: ClusterManager, cluster_mincost: ClusterLib, subtests: SubTests, request: FixtureRequest)[source]: Create and register a stake pool with pool cost >= minPoolCost.

test_stake_pool_low_cost(cluster_mincost: ClusterLib, pool_owners_pbt: list[PoolUser]) → None[source]

Try to create and register a stake pool with pool cost lower than minPoolCost.

Expect failure. Property-based test.

class cardano_node_tests.tests.test_pools.TestPoolVoteDeleg[source]

Bases: object

Tests for pool vote delegation to DRep.

get_subtests() → Generator[Callable, None, None][source]

Get pool vote delegation scenarios.

The scenarios are executed as subtests in the test_pool_delegation test.

pools(cluster_manager: ClusterManager, cluster: ClusterLib, request: FixtureRequest) → list[PoolCreationOutput][source]: Create pools for testing vote delegation scenarios.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'runs only with Tx era >= Conway'})][source]

test_pool_vote_delegation(cluster: ClusterLib, pools: list[PoolCreationOutput], subtests: SubTests)[source]: Test pool vote delegation scenarios.

class cardano_node_tests.tests.test_pools.TestStakePool[source]

Bases: object

General tests for stake pools.

test_cancel_stake_pool_deregistration(cluster_manager: ClusterManager, cluster: ClusterLib, testfile_temp_dir: Path, request: FixtureRequest)[source]

Reregister a stake pool that is in course of being retired.

Deregister stake pool in epoch + 2
Reregister the pool by resubmitting the pool registration certificate
Delegate stake address to pool again (the address is already registered)
Check that no additional pool deposit was used
Check that pool is still correctly setup
Check that the stake addresses is still delegated

test_create_stake_pool(cluster_manager: ClusterManager, cluster: ClusterLib, no_of_addr: int, request: FixtureRequest, build_method: str)[source]

Create and register a stake pool (without metadata).

Uses parametrized build method (build or build-raw) and number of pool owners (1 or 3).

Create parametrized number of pool owner addresses
Fund first pool owner address with 900 ADA
Generate pool cold keys and VRF keys
Create pool registration certificate without metadata
Submit registration certificate with pool deposit
Check that pool was registered on chain with correct parameters
Check that pool deposit was deducted from payment address

test_deregister_stake_pool(cluster_manager: ClusterManager, cluster: ClusterLib, build_method: str)[source]

Deregister stake pool.

Uses parametrized build method (build or build-raw).

Create pool with metadata and 3 pool owners
Register and delegate stake addresses to the pool
Wait for pool to be registered and receive first rewards
Generate pool deregistration certificate for future epoch
Submit deregistration certificate
Wait for retirement epoch
Check that the pool was deregistered and no longer appears in stake distribution
Check that owner stake addresses are no longer delegated to the pool
Check that the pool deposit was returned to reward account
(optional) Check pool deregistration in db-sync
(optional) Check SMASH service delisted the pool

test_pool_registration_deregistration(cluster_manager: ClusterManager, cluster: ClusterLib)[source]

Send both pool registration and deregistration certificates in single TX.

Create pool registration cert
Create pool deregistration cert
Register and deregister stake pool in single TX
Check that the pool deposit was NOT returned to reward account as the reward address is not registered (deposit is lost)

test_reregister_stake_pool(cluster_manager: ClusterManager, cluster: ClusterLib, testfile_temp_dir: Path, request: FixtureRequest)[source]

Reregister stake pool after deregistration.

Test that a deregistered pool can be re-registered by resubmitting the pool registration certificate. The stake address that was already registered does not need to pay deposit again.

Create and register pool with metadata
Delegate owner stake address to pool
Wait for retirement epoch and check pool is deregistered
Check that owner stake address is no longer delegated
Check that pool deposit was returned to reward account
Reregister the pool by resubmitting the original pool registration certificate
Delegate stake address to pool again (address is already registered, no new deposit)
Wait for pool to become active again
Check that pool was correctly registered on chain
Check that stake address was delegated to the pool
(optional) Check pool re-registration in db-sync

test_sign_in_multiple_stages(cluster_manager: ClusterManager, cluster: ClusterLib, testfile_temp_dir: Path, request: FixtureRequest)[source]

Create and register a stake pool with TX signed in multiple stages.

Create stake pool registration cert
Create witness file for each signing key
Sign TX using witness files
Create and register pool
Check that the pool was correctly registered on chain

test_stake_pool_metadata(cluster_manager: ClusterManager, cluster: ClusterLib, request: FixtureRequest, build_method: str)[source]

Create and register a stake pool with metadata.

Uses parametrized build method (build or build-raw).

Create pool metadata JSON file with name, description, ticker, homepage
Generate pool metadata hash from metadata file
Create 3 pool owner addresses
Fund first pool owner address with 900 ADA
Generate pool cold keys and VRF keys
Create pool registration certificate with metadata URL and hash
Submit registration certificate and delegate owner stake address to pool
Check that pool was registered with correct parameters
Check that stake address was delegated to pool
Check transaction view command
(optional) Check pool off-chain metadata in db-sync

test_stake_pool_not_avail_metadata(cluster_manager: ClusterManager, cluster: ClusterLib, request: FixtureRequest)[source]

Create and register a stake pool with metadata file not available.

Test that pool registration succeeds even when metadata URL points to nonexistent location.

Create pool metadata JSON file with pool details
Generate pool metadata hash from local file
Specify invalid metadata URL (file not actually hosted there)
Create pool owner address
Fund pool owner address with 900 ADA
Generate pool registration certificate with invalid metadata URL
Submit registration certificate and delegate owner stake address
Check that pool was registered despite invalid metadata URL
Check that stake address was delegated
(optional) Check db-sync records pool off-chain fetch error for invalid URL
(optional) Check SMASH service records pool error for invalid metadata

test_update_stake_pool_metadata(cluster_manager: ClusterManager, cluster: ClusterLib, build_method: str, request: FixtureRequest)[source]

Update stake pool metadata.

Register pool
Update the pool metadata by resubmitting the pool registration certificate
Check that the pool metadata hash was correctly updated on chain

test_update_stake_pool_parameters(cluster_manager: ClusterManager, cluster: ClusterLib, build_method: str, request: FixtureRequest)[source]

Update stake pool parameters.

Register pool
Update the pool parameters by resubmitting the pool registration certificate
Check that the pool parameters were correctly updated on chain

cardano_node_tests.tests.test_pools.pool_cost_start_cluster() → Path[source]: Update minPoolCost to 500.

cardano_node_tests.tests.test_protocol module

Tests for protocol state and protocol parameters.

class cardano_node_tests.tests.test_protocol.TestProtocol[source]

Bases: object

Basic tests for protocol.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'meant to run with default era or higher, where cluster era == Tx era'})][source]

test_protocol_params(cluster: ClusterLib)[source]

Check output of query protocol-parameters.

Test that protocol parameters query returns expected JSON structure with all required keys.

Execute cardano-cli query protocol-parameters command
Check that output contains all expected Conway era protocol parameter keys including: collateralPercentage, dRepDeposit, govActionDeposit, maxBlockBodySize, maxTxSize, minPoolCost, stakeAddressDeposit, stakePoolDeposit, txFeeFixed, txFeePerByte, utxoCostPerByte, and other protocol parameters
Verify no extra or missing keys in protocol parameters output

test_protocol_state_keys(cluster: ClusterLib)[source]

Check output of query protocol-state.

Test that protocol state query returns expected JSON structure with all required keys.

Execute cardano-cli query protocol-state command
Save raw output to file for debugging
Parse output as JSON
Check that output contains exactly the expected protocol state keys: candidateNonce, epochNonce, evolvingNonce, labNonce, lastEpochBlockNonce, lastSlot, oCertCounters
Verify no extra or missing keys in protocol state output

test_protocol_state_outfile(cluster: ClusterLib)[source]

Check output file produced by query protocol-state.

Test that protocol state can be written to stdout and contains expected keys.

Execute cardano-cli query protocol-state –out-file /dev/stdout command
Parse stdout output as JSON
Check that output contains exactly the expected protocol state keys
Verify no extra or missing keys in protocol state output

cardano_node_tests.tests.test_reconnect module

Tests for reconnect.

class cardano_node_tests.tests.test_reconnect.TestNodeReconnect[source]

Bases: object

Tests for nodes reconnect.

get_prometheus_metrics(port: int) → Response[source]

node_get_tip(cluster_obj: ClusterLib, node: str) → dict[str, Any][source]: Query UTxO on given node.

node_query_utxo(cluster_obj: ClusterLib, node: str, address: str = '', tx_raw_output: TxRawOutput | None = None) → list[UTXOData][source]: Query UTxO on given node.

node_submit_tx(cluster_obj: ClusterLib, node: str, temp_template: str, src_addr: AddressRecord, dst_addr: AddressRecord) → TxRawOutput[source]: Submit transaction on given node.

payment_addrs(cluster_manager: ClusterManager, cluster_singleton: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'runs only with same cluster and Tx era'}), Mark(name='skipif', args=(False,), kwargs={'reason': 'not supposed to run on long-running testnet'})][source]

test_metrics_reconnect(cluster_manager: ClusterManager, cluster_singleton: ClusterLib)[source]

Test using metrics that node reconnects after it was restarted.

Test node reconnection after restart by validating Prometheus metrics indicating successful peer connections. Performs 200 restart iterations to ensure reliability.

Get Prometheus port for pool2 from cluster configuration
For each iteration (up to 200 restarts):
- Restart pool2 node with 5 second delay
- Wait for node to sync with chain
- Fetch Prometheus metrics from node HTTP endpoint
- Parse metrics response into key-value pairs
- Check inboundGovernor_hot metric > 1 (active inbound connections)
- Check peerSelection_cold metric == 0 (no cold peers)
- Retry up to 10 times with 5 second delays if assertions fail
- Fail test if metrics don’t match after 10 attempts

test_reconnect(cluster_manager: ClusterManager, cluster_singleton: ClusterLib, payment_addrs: list[AddressRecord])[source]

Test that node reconnects after it was stopped.

Stop the node2
Submit Tx number 1 on node1
Start the stopped node2
Submit a Tx number 2 on node2
Wait for 2 new blocks
Check that node1 knows about Tx number 2, and/or node2 knows about Tx number 1

cardano_node_tests.tests.test_rollback module

Tests for rollbacks.

In rollback tests, we split the cluster into two parts. We achieve this by changing topology configuration.

class cardano_node_tests.tests.test_rollback.TestRollback[source]

Bases: object

Tests for rollbacks.

backup_topology() → Path[source]: Backup the original topology files.

node_query_utxo(cluster_obj: ClusterLib, node: str, address: str = '', tx_raw_output: TxRawOutput | None = None) → list[UTXOData][source]: Query UTxO on given node.

node_submit_tx(cluster_obj: ClusterLib, node: str, temp_template: str, src_addr: AddressRecord, dst_addr: AddressRecord) → TxRawOutput[source]: Submit transaction on given node.

node_wait_for_block(cluster_obj: ClusterLib, node: str, block_no: int) → int[source]: Wait for block number on given node.

payment_addrs(cluster_manager: ClusterManager, cluster_singleton: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

pytestmark = [Mark(name='skipif', args=(True,), kwargs={'reason': '`NUM_POOLS` must be at least 4'}), Mark(name='skipif', args=(False,), kwargs={'reason': 'runs only with same cluster and Tx era'}), Mark(name='skipif', args=(False,), kwargs={'reason': 'not supposed to run on long-running testnet'})][source]

restore_cluster(backup_topology: Path) → None[source]: Restore the original topology files == restore the cluster.

split_cluster(split_topology_dir: Path) → None[source]: Use the split topology files == split the cluster.

split_topology_dir() → Path[source]: Return path to directory with split topology files.

test_consensus_reached(cluster_manager: ClusterManager, cluster_singleton: ClusterLib, payment_addrs: list[AddressRecord], backup_topology: Path, split_topology_dir: Path)[source]

Test that global consensus is reached after rollback.

The original cluster is split into two clusters, and before securityParam number of blocks is produced, the original cluster topology gets restored.

Submit Tx number 1
Split the cluster into two separate clusters
Check that the Tx number 1 exists on both clusters
Submit a Tx number 2 on the first cluster
Check that the Tx number 2 exists only on the first cluster
Submit a Tx number 3 on the second cluster
Check that the Tx number 3 exists only on the second cluster
Restore the cluster topology
Check that global consensus was restored

test_permanent_fork(cluster_manager: ClusterManager, cluster_singleton: ClusterLib, payment_addrs: list[AddressRecord], backup_topology: Path, split_topology_dir: Path)[source]

Test that global consensus is NOT reached and the result is permanent fork.

The original cluster is split into two clusters, and after securityParam number of blocks is produced, the original cluster topology gets restored.

Submit Tx number 1
Split the cluster into two separate clusters
Submit a Tx number 2 on the first cluster
Submit a Tx number 3 on the second cluster
Wait until securityParam number of blocks is produced on both clusters
Restore the cluster topology
Check that global consensus was NOT restored

cardano_node_tests.tests.test_scripts module

Tests for multisig transactions and scripts.

Multisig
Time locking
Auxiliary scripts
Reference UTxO

class cardano_node_tests.tests.test_scripts.TestAuxiliaryScripts[source]

Bases: object

Tests for auxiliary scripts.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'runs only with Allegra+ TX'})][source]

test_tx_script_invalid(cluster: ClusterLib, payment_addrs: list[AddressRecord], build_method: str)[source]

Build transaction with invalid auxiliary script.

Expect failure.

test_tx_script_metadata_cbor(cluster: ClusterLib, payment_addrs: list[AddressRecord], build_method: str, submit_method: str)[source]

Send transaction with auxiliary script and metadata CBOR.

Check that the auxiliary script is present in the TX body.

test_tx_script_metadata_json(cluster: ClusterLib, payment_addrs: list[AddressRecord], build_method: str, submit_method: str)[source]

Send transaction with auxiliary script and metadata JSON.

Check that the auxiliary script is present in the TX body.

test_tx_script_no_metadata(cluster: ClusterLib, payment_addrs: list[AddressRecord], build_method: str, submit_method: str)[source]

Send transaction with auxiliary script and no other metadata.

Check that the auxiliary script is present in the TX body.

class cardano_node_tests.tests.test_scripts.TestBasic[source]

Bases: object

Basic tests for multisig transactions.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

test_multisig_all(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str)[source]

Send funds to and from multisig script address using “all” script type.

Test that transactions can lock funds at script address and spend them when all required signatures are provided. Uses parametrized build and submit methods.

Create multisig script requiring all payment verification keys to sign
Generate script address from multisig script
Send 2 ADA to script address from payment address
Check funds locked at script address
Spend funds from script address by providing all required signatures
Check funds transferred to destination address
(optional) Check transactions in db-sync

test_multisig_any(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str)[source]

Send funds using the any script.

Send funds to script address
Send funds from script address using single witness
Send funds from script address using multiple witnesses

test_multisig_atleast(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str)[source]

Send funds to and from multisig script address using “atLeast” script type.

Test that transactions can lock funds at script address and spend them when minimum required number of signatures is provided. Uses parametrized build and submit methods.

Create multisig script requiring at least M of N payment verification keys
Generate script address from multisig script
Send 2 ADA to script address from payment address
Check funds locked at script address
Spend funds from script address multiple times using different combinations of signatures (always meeting minimum required)
Check funds transferred to destination address
(optional) Check transactions in db-sync

test_multisig_empty_all(cluster: ClusterLib, payment_addrs: list[AddressRecord])[source]: Send funds from script address using the all script with zero skeys.

test_multisig_no_required_atleast(cluster: ClusterLib, payment_addrs: list[AddressRecord])[source]: Send funds from script address using the atLeast script with no required witnesses.

test_normal_tx_from_script_addr(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str)[source]: Send funds from script address using TX signed with skeys (not using witness files).

test_normal_tx_to_script_addr(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str)[source]: Send funds to script address using TX signed with skeys (not using witness files).

test_script_addr_length(cluster: ClusterLib, payment_addrs: list[AddressRecord])[source]

Verify that multisig script address has correct length.

Test that script address length matches standard address length (28 bytes encoded). Regression test for issue where script addresses were incorrectly 32 bytes.

Create payment verification keys for multiple addresses
Create multisig script using “all” script type with payment vkeys
Generate payment address from multisig script
Verify script address length equals standard payment address length

test_stake_keys_multisig_all(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str)[source]

Send funds to and from multisig script address using stake keys for witnessing.

Test that transactions can use stake key signatures (instead of payment keys) to witness spending from multisig script address. Uses parametrized build and submit methods.

Generate stake key pairs for multisig script witnesses
Create multisig script using stake verification keys with “all” script type
Generate payment script address from multisig script
Send 2 ADA to script address from payment address
Check funds locked at script address
Spend funds from script address by providing stake key signatures
Check funds transferred to destination address
(optional) Check transactions in db-sync

class cardano_node_tests.tests.test_scripts.TestCompatibility[source]

Bases: object

Tests for checking compatibility with previous Tx eras.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

test_auxiliary_scripts(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str)[source]: Check that it is not possible to use auxiliary script in Shelley-era Tx.

test_script_v2(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str)[source]: Check that it is not possible to use ‘SimpleScriptV2’ in Shelley-era Tx.

class cardano_node_tests.tests.test_scripts.TestDatum[source]

Bases: object

Tests for Simple Scripts V1 and V2 UTxOs with datum.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'runs only with Alonzo+ TX'})][source]

test_script_utxo_datum(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str, script_version: str)[source]: Test creating UTxO with datum on Simple Scripts V1 and V2 address.

class cardano_node_tests.tests.test_scripts.TestIncrementalSigning[source]

Bases: object

Tests for incremental signing.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

test_incremental_signing(cluster: ClusterLib, payment_addrs: list[AddressRecord], build_method: str, submit_method: str, tx_is: str)[source]

Send funds from script address using TX that is signed incrementally.

Test with Tx body created by both transaction build and transaction build-raw. Test with Tx created by both transaction sign and transaction assemble.

class cardano_node_tests.tests.test_scripts.TestNegative[source]

Bases: object

Transaction tests that are expected to fail.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

test_multisig_all_missing_skey(cluster: ClusterLib, payment_addrs: list[AddressRecord])[source]

Try to send funds from script address using the all script, omit one skey.

Expect failure.

test_multisig_any_unlisted_skey(cluster: ClusterLib, payment_addrs: list[AddressRecord])[source]

Try to send funds from script address using the any script with unlisted skey.

Expect failure.

test_multisig_atleast_low_num_of_skeys(cluster: ClusterLib, payment_addrs: list[AddressRecord])[source]

Try to send funds from script address using the atLeast script.

Num of skeys < required. Expect failure.

class cardano_node_tests.tests.test_scripts.TestNested[source]

Bases: object

Tests for nested scripts.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'runs only with Allegra+ TX'})][source]

test_invalid(cluster: ClusterLib, payment_addrs: list[AddressRecord], scenario: str, submit_method: str, build_method: str)[source]: Test scenarios where it’s NOT possible to spend from a script address.

test_nested_optional_all(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str)[source]: Check that it is possible to not meet conditions in nested “all” rule.

test_nested_script(cluster: ClusterLib, payment_addrs: list[AddressRecord], type_top: str, type_nested: str, submit_method: str, build_method: str)[source]: Check that it is possible to spend using a script with nested rules.

class cardano_node_tests.tests.test_scripts.TestReferenceUTxO[source]

Bases: object

Tests for Simple Scripts V1 and V2 on reference UTxOs.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'runs only with Babbage+ TX'})][source]

test_script_reference_utxo(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str, script_version: str)[source]: Send funds from script address where script is on reference UTxO.

test_spend_reference_script(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str, script_version: str, address_type: str)[source]

Test spending a UTxO that holds a reference script.

Create a Tx output with reference script (reference script UTxO)
Spend the reference UTxO
Check that the UTxO was spent

class cardano_node_tests.tests.test_scripts.TestTimeLocking[source]

Bases: object

Tests for time locking.

SCRIPT_AMOUNT = 2000000[source]

fund_script_after_slot_in_future(cluster: ClusterLib, payment_addrs: list[AddressRecord], request: SubRequest) → tuple[Path, str, list[UTXOData], TxRawOutput, int][source]: Create and fund script address with “after” slot in the future.

fund_script_after_slot_in_past(cluster: ClusterLib, payment_addrs: list[AddressRecord], request: SubRequest) → tuple[Path, str, list[UTXOData], TxRawOutput, int][source]: Create and fund script address with “after” slot in the past.

fund_script_before_slot_in_future(cluster: ClusterLib, payment_addrs: list[AddressRecord], request: SubRequest) → tuple[Path, str, list[UTXOData], TxRawOutput, int][source]: Create and fund script address with “before” slot in the future.

fund_script_before_slot_in_past(cluster: ClusterLib, payment_addrs: list[AddressRecord], request: SubRequest) → tuple[Path, str, list[UTXOData], TxRawOutput, int][source]: Create and fund script address with “before” slot in the past.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'runs only with Allegra+ TX'})][source]

test_after_future(cluster: ClusterLib, payment_addrs: list[AddressRecord], fund_script_after_slot_in_future: tuple[Path, str, list[UTXOData], TxRawOutput, int], request: FixtureRequest) → None[source]

Check that it’s NOT possible to spend from the script address.

The “after” slot is in the future and the given range is invalid.

test_after_past(cluster: ClusterLib, payment_addrs: list[AddressRecord], fund_script_after_slot_in_past: tuple[Path, str, list[UTXOData], TxRawOutput, int], request: FixtureRequest) → None[source]

Check that it’s NOT possible to spend from the script address.

The “after” slot is in the past.

test_before_future(cluster: ClusterLib, payment_addrs: list[AddressRecord], fund_script_before_slot_in_future: tuple[Path, str, list[UTXOData], TxRawOutput, int], request: FixtureRequest) → None[source]

Check that it’s NOT possible to spend from the script address.

The “before” slot is in the future and the given range is invalid.

test_before_past(cluster: ClusterLib, payment_addrs: list[AddressRecord], fund_script_before_slot_in_past: tuple[Path, str, list[UTXOData], TxRawOutput, int], request: FixtureRequest) → None[source]

Check that it’s NOT possible to spend from the script address.

The “before” slot is in the past.

test_script_after(cluster: ClusterLib, payment_addrs: list[AddressRecord], use_tx_validity: bool, build_method: str)[source]: Check that it is possible to spend from script address after given slot.

test_script_before(cluster: ClusterLib, payment_addrs: list[AddressRecord], use_tx_validity: bool, build_method: str)[source]: Check that it is possible to spend from script address before given slot.

test_tx_missing_validity(cluster: ClusterLib, payment_addrs: list[AddressRecord], build_method: str, slot_type: str)[source]

Check that it is NOT possible to spend from script address.

The transaction validity interval is not specified.

test_tx_negative_validity(cluster: ClusterLib, payment_addrs: list[AddressRecord], build_method: str)[source]: Check that it is NOT possible to spend from script address when validity is negative.

cardano_node_tests.tests.test_scripts.multisig_tx(cluster_obj: ClusterLib, temp_template: str, payment_address: str, dst_address: str, amount: int, payment_skey_files: list[Path], multisig_script: Path | None = None, script_utxos: list[UTXOData] | None = None, invalid_hereafter: int | None = None, invalid_before: int | None = None, submit_method: str = SubmitMethods.CLI, build_method: str = BuildMethods.BUILD_RAW) → TxRawOutput[source]: Build and submit multisig transaction.

cardano_node_tests.tests.test_smash module

Tests for basic SMASH operations.

class cardano_node_tests.tests.test_smash.TestBasicSmash[source]

Bases: object

Basic tests for SMASH service.

locked_pool(cluster_lock_pool: tuple[ClusterLib, str]) → PoolDataRecord[source]: Get id of locked pool from cluster_lock_pool fixture.

smash() → SmashClient | None[source]: Create SMASH client.

test_delist_pool(locked_pool: PoolDataRecord, smash: SmashClient, request: FixtureRequest, worker_id: str)[source]

Test delisting a pool from SMASH.

Test pool delisting functionality and verify delisted pools cannot be queried.

Get pool ID from locked pool fixture
Register cleanup function to re-enlist pool after test
Delist the pool using SMASH API
Verify delist response contains correct pool ID
Attempt to fetch metadata for delisted pool
Check that fetch fails with HTTP 403 Forbidden and “Pool is delisted” message
Add log ignore rule for expected “Delisted pool already exists!” error
Attempt to re-delist already delisted pool
Verify re-delist fails with HTTP 400 Bad Request and DbInsertError

test_enlist_pool(locked_pool: PoolDataRecord, smash: SmashClient)[source]

Test enlisting a pool in SMASH.

Test pool enlisting functionality after delisting and verify enlisted pools can be queried.

Get pool ID from locked pool fixture
Attempt to enlist already enlisted pool
Verify enlist fails with HTTP 404 Not Found and “RecordDoesNotExist” error
Delist the pool to prepare for enlist test
Verify delisted pool cannot be queried (HTTP 403 Forbidden)
Enlist the delisted pool
Verify enlist response contains correct pool ID
Fetch metadata for newly enlisted pool
Verify metadata retrieval succeeds for enlisted pool

test_fetch_pool_metadata(locked_pool: PoolDataRecord, smash: SmashClient)[source]

Test fetching pool metadata from SMASH.

Test retrieval of off-chain pool metadata from SMASH server and verify it matches db-sync records.

Get pool ID from locked pool fixture
Query off-chain pool metadata from db-sync (retry with timeout for availability)
Extract expected metadata fields (name, description, ticker, homepage)
Fetch pool metadata from SMASH using pool ID and metadata hash
Verify SMASH metadata matches db-sync metadata exactly

test_reserve_ticker(cluster: ClusterLib, smash: SmashClient, request: FixtureRequest)[source]

Test reserving a ticker for a pool in SMASH.

Test ticker reservation functionality and verify duplicate reservations are rejected.

Select random pool ID from cluster
Register cleanup function to delete reserved tickers after test completion
Reserve 3-character random ticker for the pool
Verify reservation response contains correct ticker name
Attempt to reserve already-taken ticker for same pool
Verify re-reservation fails with HTTP 400 Bad Request and TickerAlreadyReserved error

cardano_node_tests.tests.test_smash.check_request_error(err: RequestException, expected_status: HTTPStatus, expected_code: str | None, expected_description: str) → None[source]: Assert expected HTTP errors in requests, handling both JSON and text responses.

cardano_node_tests.tests.test_socket_path module

Tests for –socket-path CLI argument.

class cardano_node_tests.tests.test_socket_path.TestNegativeSocketPath[source]

Bases: object

Negative tests for cardano-cli … –socket-path.

test_neg_build_transfer_funds(set_socket_path: None, cluster: ClusterLib, payment_addrs: list[AddressRecord], env_scenario: str, socket_scenario: str)[source]

Build transfer transaction with incorrect socket path configuration.

Test parametrized scenarios where CARDANO_NODE_SOCKET_PATH env variable and/or –socket-path argument are missing or contain wrong values.

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

Create 2 payment addresses and fund first address
Set CARDANO_NODE_SOCKET_PATH to wrong path or remove it (env_scenario parameter)
Set –socket-path to wrong path or remove it (socket_scenario parameter)
Attempt to build transaction sending 1.5 ADA from source to destination address
Check that transaction build fails with expected error message
Verify error is “Network.Socket.connect” when socket path is wrong
Verify error is “Missing: –socket-path SOCKET_PATH” when both missing

Expect failure.

test_neg_query_pool_state(set_socket_path: None, cluster: ClusterLib, env_scenario: str, socket_scenario: str)[source]

Query pool state with incorrect socket path configuration.

Test parametrized scenarios where CARDANO_NODE_SOCKET_PATH env variable and/or –socket-path argument are missing or contain wrong values.

Set CARDANO_NODE_SOCKET_PATH to wrong path or remove it (env_scenario parameter)
Set –socket-path to wrong path or remove it (socket_scenario parameter)
Execute cardano-cli query pool-state command for specific pool ID
Check that command fails with expected error message
Verify error is “Network.Socket.connect” when socket path is wrong
Verify error is “Missing: –socket-path SOCKET_PATH” when both missing

Expect failure.

test_neg_query_protocol_params(set_socket_path: None, cluster: ClusterLib, env_scenario: str, socket_scenario: str)[source]

Query protocol parameters with incorrect socket path configuration.

Test parametrized scenarios where CARDANO_NODE_SOCKET_PATH env variable and/or –socket-path argument are missing or contain wrong values.

Set CARDANO_NODE_SOCKET_PATH to wrong path or remove it (env_scenario parameter)
Set –socket-path to wrong path or remove it (socket_scenario parameter)
Execute cardano-cli query protocol-parameters command
Check that command fails with expected error message
Verify error is “Network.Socket.connect” when socket path is wrong
Verify error is “Missing: –socket-path SOCKET_PATH” when both missing

Expect failure.

test_neg_query_protocol_state(set_socket_path: None, cluster: ClusterLib, env_scenario: str, socket_scenario: str)[source]

Query protocol state with incorrect socket path configuration.

Test parametrized scenarios where CARDANO_NODE_SOCKET_PATH env variable and/or –socket-path argument are missing or contain wrong values.

Set CARDANO_NODE_SOCKET_PATH to wrong path or remove it (env_scenario parameter)
Set –socket-path to wrong path or remove it (socket_scenario parameter)
Execute cardano-cli query protocol-state command
Check that command fails with expected error message
Verify error is “Network.Socket.connect” when socket path is wrong
Verify error is “Missing: –socket-path SOCKET_PATH” when both missing

Expect failure.

test_neg_query_stake_addr_info(set_socket_path: None, cluster: ClusterLib, env_scenario: str, socket_scenario: str)[source]

Query stake address info with incorrect socket path configuration.

Test parametrized scenarios where CARDANO_NODE_SOCKET_PATH env variable and/or –socket-path argument are missing or contain wrong values.

Set CARDANO_NODE_SOCKET_PATH to wrong path or remove it (env_scenario parameter)
Set –socket-path to wrong path or remove it (socket_scenario parameter)
Execute cardano-cli query stake-address-info command for specific stake address
Check that command fails with expected error message
Verify error is “Network.Socket.connect” when socket path is wrong
Verify error is “Missing: –socket-path SOCKET_PATH” when both missing

Expect failure.

test_neg_query_stake_distribution(set_socket_path: None, cluster: ClusterLib, env_scenario: str, socket_scenario: str)[source]

Query stake distribution with incorrect socket path configuration.

Test parametrized scenarios where CARDANO_NODE_SOCKET_PATH env variable and/or –socket-path argument are missing or contain wrong values.

Set CARDANO_NODE_SOCKET_PATH to wrong path or remove it (env_scenario parameter)
Set –socket-path to wrong path or remove it (socket_scenario parameter)
Execute cardano-cli query stake-distribution command
Check that command fails with expected error message
Verify error is “Network.Socket.connect” when socket path is wrong
Verify error is “Missing: –socket-path SOCKET_PATH” when both missing

Expect failure.

class cardano_node_tests.tests.test_socket_path.TestSocketPath[source]

Bases: object

Tests for cardano-cli … –socket-path.

test_build_transfer_funds(set_socket_path: None, cluster: ClusterLib, payment_addrs: list[AddressRecord], env_scenario: str)[source]

Send funds to payment address.

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

Send funds from 1 source address to 1 destination address
Check expected balances for both source and destination addresses

test_query_pool_state(set_socket_path: None, cluster: ClusterLib, env_scenario: str)[source]

Query pool state using –socket-path CLI argument.

Test that cardano-cli can query pool state when socket path is specified via –socket-path argument instead of CARDANO_NODE_SOCKET_PATH env variable.

Unset CARDANO_NODE_SOCKET_PATH environment variable
Set socket path via –socket-path CLI argument
Optionally set wrong or remove socket path (env_scenario parameter)
Execute cardano-cli query pool-state command for specific pool ID
Verify query succeeds with –socket-path argument

test_query_protocol_params(set_socket_path: None, cluster: ClusterLib, env_scenario: str)[source]

Query protocol parameters using –socket-path CLI argument.

Test that cardano-cli can query protocol parameters when socket path is specified via –socket-path argument instead of CARDANO_NODE_SOCKET_PATH env variable.

Unset CARDANO_NODE_SOCKET_PATH environment variable
Set socket path via –socket-path CLI argument
Optionally set wrong or remove socket path (env_scenario parameter)
Execute cardano-cli query protocol-parameters command
Check that parameters contain protocolVersion field
Verify query succeeds with –socket-path argument

test_query_protocol_state(set_socket_path: None, cluster: ClusterLib, env_scenario: str)[source]

Query protocol state using –socket-path CLI argument.

Test that cardano-cli can query protocol state when socket path is specified via –socket-path argument instead of CARDANO_NODE_SOCKET_PATH env variable.

Unset CARDANO_NODE_SOCKET_PATH environment variable
Set socket path via –socket-path CLI argument
Optionally set wrong or remove socket path (env_scenario parameter)
Execute cardano-cli query protocol-state command
Check that protocol state contains lastSlot field
Verify query succeeds with –socket-path argument

test_query_stake_addr_info(set_socket_path: None, cluster: ClusterLib, env_scenario: str)[source]

Query stake address info using –socket-path CLI argument.

Test that cardano-cli can query stake address info when socket path is specified via –socket-path argument instead of CARDANO_NODE_SOCKET_PATH env variable.

Unset CARDANO_NODE_SOCKET_PATH environment variable
Set socket path via –socket-path CLI argument
Optionally set wrong or remove socket path (env_scenario parameter)
Execute cardano-cli query stake-address-info command for specific stake address
Verify query succeeds with –socket-path argument

test_query_stake_distribution(set_socket_path: None, cluster: ClusterLib, env_scenario: str)[source]

Query stake distribution using –socket-path CLI argument.

Test that cardano-cli can query stake distribution when socket path is specified via –socket-path argument instead of CARDANO_NODE_SOCKET_PATH env variable.

Unset CARDANO_NODE_SOCKET_PATH environment variable
Set socket path via –socket-path CLI argument
Optionally set wrong or remove socket path (env_scenario parameter)
Execute cardano-cli query stake-distribution command
Check that distribution contains pool IDs starting with “pool”
Verify query succeeds with –socket-path argument

cardano_node_tests.tests.test_socket_path.payment_addrs(set_socket_path: None, cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

cardano_node_tests.tests.test_socket_path.set_socket_path(cluster_manager: ClusterManager, cluster: ClusterLib) → Generator[None, None, None][source]: Unset CARDANO_NODE_SOCKET_PATH and set path for cardano-cli … –socket-path.

cardano_node_tests.tests.test_staking_no_rewards module

Tests for checking staking scenarios where no rewards are expected.

class cardano_node_tests.tests.test_staking_no_rewards.TestNoRewards[source]

Bases: object

test_deregister_reward_addr_retire_pool(cluster_manager: ClusterManager, cluster_lock_pool_use_rewards: tuple[ClusterLib, str])[source]

Test deregistering reward address and retiring stake pool.

The pool deposit is lost when reward address is deregistered before the pool is retired.

Wait for first reward for the pool
Withdraw pool rewards to payment address
Deregister the pool reward address
Check that the key deposit was returned
Check that pool owner is NOT receiving rewards
Deregister stake pool
Check that the pool deposit was NOT returned to reward or stake address
Return the pool to the original state - reregister the pool, register the reward address, delegate the stake address to the pool
Check that pool deposit was needed
Check that pool owner is receiving rewards

test_no_reward_deregistered_reward_addr(cluster_manager: ClusterManager, cluster_lock_pool_use_rewards: tuple[ClusterLib, str])[source]

Check that the reward address is not receiving rewards when deregistered.

The stake pool continues to operate normally and those who delegate to that pool receive rewards.

Delegate stake address
Wait for first reward
Withdraw pool rewards to payment address
Deregister the pool reward address
Check that the key deposit was returned
Check that pool owner is NOT receiving rewards
Check that new rewards are received by those delegating to the pool
Return the pool to the original state - reregister reward address
Check that pool owner is receiving rewards

test_no_reward_deregistered_stake_addr(cluster_manager: ClusterManager, cluster_lock_pool_use_rewards: tuple[ClusterLib, str])[source]

Check that the pool is not receiving rewards when owner’s stake address is deregistered.

When the owner’s stake address is deregistered (i.e. owner’s stake is lower than pledge), neither pool owners nor those who delegate to that pool receive rewards.

Delegate stake address
Wait for first reward
Deregister stake address - owner’s stake is lower than pledge
Check that the key deposit was returned
Check that NO new rewards were received by those delegating to the pool
Check that pool owner is also NOT receiving rewards
Return the pool to the original state - reregister stake address and delegate it to the pool
Check that new rewards were received by those delegating to the pool
Check that pool owner is also receiving rewards

test_no_reward_unmet_pledge1(cluster_manager: ClusterManager, cluster_lock_pool: tuple[ClusterLib, str])[source]

Check that the stake pool is not receiving rewards when pledge is not met.

When the pledge is higher than available funds, neither pool owners nor those who delegate to that pool receive rewards.

Delegate stake address
Wait for first reward
Increase the needed pledge amount - update the pool parameters by resubmitting the pool registration certificate - the funds are now lower than what is needed by the stake pool
Check that NO new rewards were received by those delegating to the pool
Check that pool owner is also NOT receiving rewards
Return the pool to the original state - restore pledge settings
Check that new rewards were received by those delegating to the pool
Check that pool owner is also receiving rewards

test_no_reward_unmet_pledge2(cluster_manager: ClusterManager, cluster_lock_pool: tuple[ClusterLib, str])[source]

Check that the stake pool is not receiving rewards when pledge is not met.

When the pledge is higher than available funds, neither pool owners nor those who delegate to that pool receive rewards.

Delegate stake address
Wait for first reward
Withdraw part of the pledge - the funds are lower than what is needed by the stake pool
Check that NO new rewards were received by those delegating to the pool
Check that pool owner is also NOT receiving rewards
Return the pool to the original state - restore pledge funds
Check that new rewards were received by those delegating to the pool
Check that pool owner is also receiving rewards

cardano_node_tests.tests.test_staking_no_rewards.cluster_lock_pool_use_rewards(cluster_manager: ClusterManager) → tuple[ClusterLib, str][source]: Lock any pool, use pots, and return instance of clusterlib.ClusterLib.

cardano_node_tests.tests.test_staking_rewards module

Tests for staking, rewards, blocks production on real block-producing pools.

class cardano_node_tests.tests.test_staking_rewards.RewardRecord(epoch_no: int, reward_total: int, reward_per_epoch: int, member_pool_id: str = '', leader_pool_ids: list[str] | tuple = (), stake_total: int = 0)[source]

Bases: object

epoch_no: int[source]

leader_pool_ids: list[str] | tuple = ()[source]

member_pool_id: str = ''[source]

reward_per_epoch: int[source]

reward_total: int[source]

stake_total: int = 0[source]

class cardano_node_tests.tests.test_staking_rewards.TestNegativeWithdrawal[source]

Bases: object

Tests for rewards withdrawal that are expected to fail.

pool_users(cluster_manager: ClusterManager, cluster_use_pool: tuple[ClusterLib, str]) → tuple[PoolUser, PoolUser][source]

test_withdrawal_wrong_amount(cluster_use_pool: tuple[ClusterLib, str], pool_users: tuple[PoolUser, PoolUser]) → None[source]

Test that it is not possible to withdraw other amount than the total reward amount.

Expect failure. Property-based test.

class cardano_node_tests.tests.test_staking_rewards.TestRewards[source]

Bases: object

Tests for checking expected rewards.

test_2_pools_same_reward_addr(cluster_manager: ClusterManager, cluster_lock_two_pools: tuple[ClusterLib, str, str])[source]

Check that one reward address used for two pools receives rewards for both of them.

Set pool2 reward address to the reward address of pool1 by resubmitting the pool registration certificate
Collect data for both pool1 and pool2 for several epochs and with the help of db-sync
- Check that the original reward address for pool2 is NOT receiving rewards
- Check that the reward address for pool1 is now receiving rewards for both pools
Check records in db-sync
- Transaction inputs, outputs, withdrawals, etc.
- Reward amounts received each epoch
- Expected pool ids

test_decreasing_reward_transferred_funds(cluster_manager: ClusterManager, cluster_use_pool_and_rewards: tuple[ClusterLib, str])[source]

Check that rewards are gradually decreasing when funds are being transferred.

Even though nothing is staked and rewards are being transferred from reward address, there are still some funds staked on the reward address at the time ledger snapshot is taken. For that reason the reward amount received every epoch is gradually decreasing over the period of several epochs until it is finally 0.

Delegate stake address
Wait for first reward
Transfer all funds from payment address back to faucet, so no funds are staked
Keep withdrawing new rewards so reward balance is 0
Check that reward amount is decreasing epoch after epoch

test_redelegation(cluster_manager: ClusterManager, cluster_use_two_pools_and_rewards: tuple[ClusterLib, str, str])[source]

Check rewards received by stake address over multiple epochs.

The address is re-delegated and deregistred / re-registered multiple times.

Delegate stake address to pool
In next epoch, re-delegate to another pool
In next epoch, deregister stake address, immediately re-register and delegate to pool
In next epoch, deregister stake address, wait for second half of an epoch, re-register and delegate to pool
While doing the steps above, collect data for pool user for 8 epochs
- Each epoch check ledger state (expected data in pstake*, delegation, stake amount)
- Each epoch check received reward with reward in ledger state
(optional) Check records in db-sync

test_reward_addr_delegation(cluster_manager: ClusterManager, cluster_lock_pool_and_pots: tuple[ClusterLib, str])[source]

Check that the rewards address can be delegated and receive rewards.

Tests https://github.com/IntersectMBO/cardano-node/issues/1964

The pool has a reward address that is different from pool owner’s stake address.

Delegate reward address to the pool
Collect reward address data for 8 epochs and
- Each epoch check ledger state (expected data in pstake*, delegation, stake amount)
- Each epoch check received reward with reward in ledger state
- Check that reward address receives rewards for its staked amount + the pool owner’s pledge (and pool cost)
- Send TXs with MIR certs that transfer funds from reserves and treasury to pool reward address and check the reward was received as expected
Check records in db-sync
- Transaction inputs, outputs, withdrawals, etc.
- Reward amounts received each epoch
- Expected pool id
- Expected reward types

test_reward_amount(cluster_manager: ClusterManager, cluster_use_pool_and_rewards: tuple[ClusterLib, str])[source]

Check that the stake address and pool owner are receiving rewards.

Create two payment addresses that share single stake address
Register and delegate the stake address to pool
Create UTxOs with native tokens
Collect data for pool owner and pool users for 9 epochs
- Each epoch check ledger state (expected data in pstake*, delegation, stake amount)
- Each epoch check received reward with reward in ledger state
Withdraw rewards to payment address
Burn native tokens
(optional) Check records in db-sync

test_reward_simple(cluster_manager: ClusterManager, cluster_and_pool: tuple[ClusterLib, str])[source]

Check that the stake address and pool owner are receiving rewards.

Delegate to pool
Wait for rewards for pool owner and pool users for up to 4 epochs
Withdraw rewards to payment address

cardano_node_tests.tests.test_staking_rewards.cluster_and_pool(cluster_manager: ClusterManager) → tuple[ClusterLib, str][source]

cardano_node_tests.tests.test_staking_rewards.cluster_lock_pool_and_pots(cluster_manager: ClusterManager) → tuple[ClusterLib, str][source]

cardano_node_tests.tests.test_staking_rewards.cluster_lock_two_pools(cluster_manager: ClusterManager) → tuple[ClusterLib, str, str][source]

cardano_node_tests.tests.test_staking_rewards.cluster_use_pool_and_rewards(cluster_manager: ClusterManager) → tuple[ClusterLib, str][source]: Mark any pool and all pots as “in use” and return instance of clusterlib.ClusterLib.

cardano_node_tests.tests.test_staking_rewards.cluster_use_two_pools_and_rewards(cluster_manager: ClusterManager) → tuple[ClusterLib, str, str][source]

cardano_node_tests.tests.test_tx_basic module

Tests for basic transactions.

class cardano_node_tests.tests.test_tx_basic.TestBasicTransactions[source]

Bases: object

Test basic transactions - transferring funds, transaction IDs.

byron_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create 2 new Byron payment addresses.

cluster_default_tx_era(cluster: ClusterLib, request: SubRequest) → ClusterLib[source]

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create 2 new payment addresses.

payment_addrs_disposable(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create 2 new payment addresses.

payment_addrs_no_change(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create 2 new payment addresses for test_build_no_change.

test_build_multiple_same_txins(cluster: ClusterLib, payment_addrs: list[AddressRecord])[source]

Build a transaction with multiple identical txins.

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

test_build_no_change(cluster: ClusterLib, payment_addrs_no_change: list[AddressRecord], submit_method: str)[source]

Send funds to payment address and balance the outputs so that there is no change.

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

Try to build a Tx that transfers all available funds, and extract fee amount from the error message
Transfer all available funds minus fee from source address to destination address
Check that no change UTxO was created
(optional) Check transactions in db-sync

test_byron_fee_too_small(cluster: ClusterLib, payment_addrs: list[AddressRecord], byron_addrs: list[AddressRecord], submit_method: str)[source]

Test cardano-node issue #4752.

Use cardano-cli transaction build command for building a transaction that needs to be signed by Byron skey. Check if calculated fee is too small and if Tx submit fails.

test_default_tx_era(cluster: ClusterLib, cluster_default_tx_era: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str)[source]

Test default Tx era.

Check that default Tx era is implicit as latest
Check that default Tx era can be specified explicitly by its name

test_duplicate_signing_keys(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str)[source]

Send a transaction with duplicate signing key.

Check that it is possible to specify the same signing key twice.

Create transaction from source address to destination address with 2 ADA
Sign transaction by specifying the same required signing key twice
Submit the transaction and verify it succeeds
Check expected balances for both source and destination addresses
(optional) Check transactions in db-sync

test_extra_signing_keys(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str)[source]

Send a transaction with an extra unnecessary signing key.

Check that it is possible to use unneeded signing key in addition to the necessary signing keys for signing the transaction.

Create transaction from source address to destination address with 2 ADA
Sign transaction with both required signing key and an extra unneeded signing key
Submit the transaction and verify it succeeds
Check expected balances for both source and destination addresses
(optional) Check transactions in db-sync

test_far_future_ttl(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str)[source]

Send a transaction with ttl (time-to-live) far in the future.

Uses cardano-cli transaction build-raw command.

Create transaction from source address to destination address with 2 ADA
Set TTL to current slot + 10,000,000 slots (far in the future)
Build, sign and submit the transaction
Verify the transaction succeeds with far-future TTL
(optional) Check transactions in db-sync

test_funds_to_valid_address(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str)[source]

Send funds to a valid payment address.

The destination address is a valid address that was generated sometime in the past. The test verifies it is possible to use a valid address even though it was not generated while running a specific cardano network.

Send funds from 1 source address to 1 destination address
Check expected balances for both source and destination addresses
Check min UTxO value
(optional) Check transactions in db-sync

test_get_txid(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str, build_method: str)[source]

Get transaction ID (txid) from transaction body.

Transaction ID is a hash of transaction body and doesn’t change for a signed TX.

Send funds from 1 source address to 1 destination address
Get txid from transaction body
Get txid from signed transaction
Check that txid from transaction body matches the txid from signed transaction
Check that txid has expected length
Check that the txid is listed in UTxO hashes for both source and destination addresses
(optional) Check transactions in db-sync

test_missing_ttl(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str)[source]

Submit a transaction with a missing –ttl (–invalid-hereafter) parameter.

Uses cardano-cli transaction build-raw command without specifying TTL.

Build transaction using build-raw without –invalid-hereafter parameter
Sign the transaction
Submit the transaction and verify it succeeds
Check expected balance for source address
(optional) Check transactions in db-sync

test_missing_tx_out(cluster: ClusterLib, payment_addrs: list[AddressRecord])[source]

Build a transaction with a missing –tx-out parameter.

Uses cardano-cli transaction build-raw command.

Prepare transaction inputs with –tx-in from source address
Build transaction using build-raw without any –tx-out parameter
Verify that transaction can be built (no output validation at build-raw stage)

test_multiple_same_txins(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str)[source]

Build and submit a transaction with multiple identical txins.

Uses cardano-cli transaction build-raw command.

Prepare transaction with the same txin specified multiple times
Build transaction using build-raw with duplicate –tx-in parameters
Sign the transaction
Submit the transaction and verify it succeeds
Check expected balance for source address
(optional) Check transactions in db-sync

test_no_txout(cluster: ClusterLib, cluster_manager: ClusterManager, submit_method: str)[source]

Send transaction with just fee, no UTxO is produced.

Submit a transaction where all funds available on source address is used for fee
Check that no UTxOs are created by the transaction
Check that there are no funds left on source address
(optional) Check transactions in db-sync

test_sign_wrong_file(cluster: ClusterLib, payment_addrs: list[AddressRecord], file_type: str, submit_method: str)[source]

Sign other file type than the one specified by command line option (Tx vs Tx body).

Specify Tx file and pass Tx body file
Specify Tx body file and pass Tx file
(optional) Check transactions in db-sync

test_transfer_all_build_estimate(cluster: ClusterLib, payment_addrs_disposable: list[AddressRecord])[source]

Transfer all funds from one payment address to another.

Use the transaction build-estimate command.

Transfer all available funds from 1 source address to 1 destination address
Check expected balance for destination addresses
Check that balance for source address is 0 Lovelace
Check output of the transaction view command
(optional) Check transactions in db-sync

test_transfer_all_funds(cluster: ClusterLib, payment_addrs_disposable: list[AddressRecord], submit_method: str)[source]

Send ALL funds from one payment address to another.

Send all available funds from 1 source address to 1 destination address
Check expected balance for destination addresses
Check that balance for source address is 0 Lovelace
Check output of the transaction view command
(optional) Check transactions in db-sync

test_transfer_funds(cluster: ClusterLib, payment_addrs: list[AddressRecord], byron_addrs: list[AddressRecord], src_addr_type: str, dst_addr_type: str, amount: int, submit_method: str, build_method: str)[source]

Send funds to payment address.

Send funds from 1 source address to 1 destination address
Check expected balances for both source and destination addresses
(optional) Check transactions in db-sync

test_transfer_some_build_estimate(cluster: ClusterLib, payment_addrs: list[AddressRecord])[source]

Transfer some funds from one payment address to another.

Use the transaction build-estimate command.

Transfer some available funds from 1 source address to 1 destination address
Check expected balance for source addresses
Check expected balance for destination addresses
Check output of the transaction view command
(optional) Check transactions in db-sync

test_utxo_with_datum_hash(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str)[source]

Create a UTxO with datum hash in a regular address and spend it.

Create a UTxO with a datum hash at the payment address
Check that the UTxO was created with the respective datum hash
Spend the UTxO (not providing the datum hash)
Check that the UTxO was spent
(optional) Check transactions in db-sync

class cardano_node_tests.tests.test_tx_basic.TestIncrementalSigning[source]

Bases: object

Test incremental signing of transactions.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

test_incremental_signing(cluster: ClusterLib, payment_addrs: list[AddressRecord], build_method: str, tx_is: str, submit_method: str)[source]

Test sending funds using Tx that is signed incrementally.

Test with Tx body created by both transaction build and transaction build-raw. Test with Tx created by both transaction sign and transaction assemble.

Create a transaction
Sign the transaction incrementally with part of the signing keys
Sign the transaction incrementally with the rest of the signing keys, except of the required one
Sign the transaction multiple times with the same skey to see that it doesn’t affect the Tx fee
Check that the transaction cannot be submitted
Sign the transaction with the required signing key
Check that the transaction can be submitted
Check expected balances for both source and destination addresses
(optional) Check transactions in db-sync

class cardano_node_tests.tests.test_tx_basic.TestMultiInOut[source]

Bases: object

Test transactions with multiple txins and/or txouts.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create 201 new payment addresses.

test_10_transactions(cluster: ClusterLib, payment_addrs: list[AddressRecord], submit_method: str)[source]

Send 10 transactions to payment address.

Send funds from 1 source address to 1 destination address in 10 separate transactions
Check expected balances for both source and destination addresses
(optional) Check transactions in db-sync

test_transaction_to_100_addrs_from_50_addrs(cluster: ClusterLib, payment_addrs: list[AddressRecord], amount: int, submit_method: str, build_method: str)[source]

Test 1 transaction from 50 payment addresses to 100 payment addresses.

Send funds from 50 source addresses to 100 destination addresses
Check expected balances for both source and destination addresses
(optional) Check transactions in db-sync

test_transaction_to_10_addrs_from_10_addrs(cluster: ClusterLib, payment_addrs: list[AddressRecord], amount: int, submit_method: str, build_method: str)[source]

Test 1 transaction from 10 payment addresses to 10 payment addresses.

Send funds from 10 source addresses to 10 destination addresses
Check expected balances for both source and destination addresses
(optional) Check transactions in db-sync

test_transaction_to_10_addrs_from_1_addr(cluster: ClusterLib, payment_addrs: list[AddressRecord], amount: int, submit_method: str, build_method: str)[source]

Test 1 transaction from 1 payment address to 10 payment addresses.

Send funds from 1 source address to 10 destination addresses
Check expected balances for both source and destination addresses
(optional) Check transactions in db-sync

test_transaction_to_1_addr_from_10_addrs(cluster: ClusterLib, payment_addrs: list[AddressRecord], amount: int, submit_method: str, build_method: str)[source]

Test 1 transaction from 10 payment addresses to 1 payment address.

Send funds from 10 source addresses to 1 destination address
Check expected balances for both source and destination addresses
(optional) Check transactions in db-sync

cardano_node_tests.tests.test_tx_fees module

Tests for fees of various kinds of transactions.

class cardano_node_tests.tests.test_tx_fees.TestExpectedFees[source]

Bases: object

Test expected fees.

pool_users(cluster_manager: ClusterManager, cluster: ClusterLib) → list[PoolUser][source]: Create pool users.

test_addr_deregistration_fees(cluster: ClusterLib, pool_users: list[PoolUser], addr_fee: tuple[int, int])[source]

Test stake address deregistration fees for various numbers of addresses.

Test with 1, 3, 5, or 10 stake addresses.

Select specified number of pool users
Generate stake address deregistration certificates with deposit refund amounts
Calculate transaction fee for deregistering specified number of stake addresses
Verify fee matches expected value within acceptable interval

test_addr_registration_fees(cluster: ClusterLib, pool_users: list[PoolUser], addr_fee: tuple[int, int])[source]

Test stake address registration fees for various numbers of addresses.

Test with 1, 3, 5, or 10 stake addresses.

Select specified number of pool users
Generate stake address registration certificates with deposit amounts
Calculate transaction fee for registering specified number of stake addresses
Verify fee matches expected value within acceptable interval

test_pool_deregistration_fees(cluster: ClusterLib, pool_users: list[PoolUser], addr_fee: tuple[int, int])[source]

Test pool deregistration fees for various numbers of pool owners.

Test with 1, 3, 5, or 10 pool owner addresses.

Create pool metadata and cold key pair for the pool
Select specified number of pool owners
Generate pool deregistration certificate for next epoch
Calculate transaction fee for pool deregistration with specified number of owners
Verify fee matches expected value within acceptable interval

test_pool_registration_fees(cluster: ClusterLib, pool_users: list[PoolUser], addr_fee: tuple[int, int])[source]

Test pool registration fees for various numbers of pool owners.

Test with 1, 3, 5, or 10 pool owner addresses.

Create VRF key pair and cold key pair for the pool node
Generate stake address registration certificates for pool owners
Generate stake address delegation certificates delegating to the pool
Generate pool registration certificate with metadata
Calculate transaction fee for pool registration with specified number of owners
Verify fee matches expected value within acceptable interval

test_transaction_to_100_addrs_from_100_addrs_fees(cluster: ClusterLib, pool_users: list[PoolUser], amount_expected: tuple[int, int])[source]

Test fees for 1 tx from 100 payment addresses to 100 payment addresses.

Test with amounts of 1, 100, 11,000, or 100,000 Lovelace per destination.

Prepare transaction inputs from 100 source addresses
Prepare transaction outputs to 100 destination addresses with specified amount each
Calculate transaction fee
Verify fee matches expected value within acceptable interval

test_transaction_to_10_addrs_from_10_addrs_fees(cluster: ClusterLib, pool_users: list[PoolUser], amount_expected: tuple[int, int])[source]

Test fees for 1 tx from 10 payment addresses to 10 payment addresses.

Test with amounts of 1, 100, 11,000, or 100,000 Lovelace per destination.

Prepare transaction inputs from 10 source addresses
Prepare transaction outputs to 10 destination addresses with specified amount each
Calculate transaction fee
Verify fee matches expected value within acceptable interval

test_transaction_to_10_addrs_from_1_addr_fees(cluster: ClusterLib, pool_users: list[PoolUser], amount_expected: tuple[int, int])[source]

Test fees for 1 tx from 1 payment address to 10 payment addresses.

Test with amounts of 1, 100, 11,000, or 100,000 Lovelace per destination.

Prepare transaction inputs from 1 source address
Prepare transaction outputs to 10 destination addresses with specified amount each
Calculate transaction fee
Verify fee matches expected value within acceptable interval

test_transaction_to_1_addr_from_10_addrs_fees(cluster: ClusterLib, pool_users: list[PoolUser], amount_expected: tuple[int, int])[source]

Test fees for 1 tx from 10 payment addresses to 1 payment address.

Test with amounts of 1, 100, 11,000, or 100,000 Lovelace.

Prepare transaction inputs from 10 source addresses
Prepare transaction output to 1 destination address with specified amount
Calculate transaction fee
Verify fee matches expected value within acceptable interval

test_transaction_to_1_addr_from_1_addr_fees(cluster: ClusterLib, pool_users: list[PoolUser], amount_expected: tuple[int, int])[source]

Test fees for 1 tx from 1 payment address to 1 payment address.

Test with amounts of 1, 100, 11,000, or 100,000 Lovelace.

Prepare transaction inputs from source address
Prepare transaction outputs to destination address with specified amount
Calculate transaction fee
Verify fee matches expected value within acceptable interval

class cardano_node_tests.tests.test_tx_fees.TestFee[source]

Bases: object

General fees tests.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create 2 new payment addresses.

test_expected_or_higher_fee(cluster: ClusterLib, payment_addrs: list[AddressRecord], fee_add: int)[source]

Send a transaction with fee that is same or higher than expected.

Test with exact fee, fee + 1,000, fee + 100,000, and fee + 1,000,000 Lovelace.

Calculate expected fee for transaction sending 2 ADA
Create transaction with fee equal to or higher than calculated fee
Send transaction and verify it succeeds
Check that actual fee matches specified fee
Check expected balances for both source and destination addresses

test_negative_fee(cluster: ClusterLib, payment_addrs: list[AddressRecord]) → None[source]

Try to send a transaction with negative fee (property-based test).

Expect failure.

Uses hypothesis property-based testing to generate negative fee values.

Create transaction from source address to destination address with 10 Lovelace
Attempt to send transaction with negative fee value
Check that transaction building fails with CLI error about fee parameter

test_smaller_fee(cluster: ClusterLib, payment_addrs: list[AddressRecord], fee_change: float)[source]

Try to send a transaction with smaller-than-expected fee.

Expect failure.

Test with fee that is 0x, 1/1.1x, 1/1.5x, or 1/2x of the calculated fee.

Calculate expected fee for transaction sending 10 Lovelace
Create transaction with fee reduced by parametrized factor
Attempt to send transaction with insufficient fee
Check that transaction submission fails with FeeTooSmallUTxO error

cardano_node_tests.tests.test_tx_many_utxos module

Tests for transactions with many UTxOs.

class cardano_node_tests.tests.test_tx_many_utxos.TestManyUTXOs[source]

Bases: object

Test transaction with many UTxOs and small amounts of Lovelace.

cluster(cluster_manager: ClusterManager) → ClusterLib[source]

many_utxos(cluster: ClusterLib, payment_addrs: list[AddressRecord]) → tuple[AddressRecord, AddressRecord][source]: Generate many UTxOs (100000+) with 1-2 ADA.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create new payment addresses.

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'expensive test, skip when cluster era is different from TX era'})][source]

test_mini_transactions(cluster: ClusterLib, many_utxos: tuple[AddressRecord, AddressRecord], subtests: SubTests)[source]

Test transactions with many UTxOs (350+) containing small amounts of ADA.

Tests blockchain performance under stress with transactions containing hundreds of inputs (UTxOs) with varying amounts. Uses cardano-cli transaction build-raw command.

Use pre-funded source address with 100000+ UTxOs containing 1-10 ADA each
Use destination address with 100000+ UTxOs
Query and sort source UTxOs by amount
Select 350 UTxOs as transaction inputs (mix of small and large amounts)
Calculate transaction fee with selected inputs
Optimize input UTxO selection to minimize total funds while covering amount + fee
Build raw transaction with 350+ inputs sending 1.5/5/10 ADA to destination
Sign and submit transaction
Check that source address balance decreased by sent amount + fee
Check that destination address balance increased by sent amount
(optional) Check transaction records in db-sync
Repeat test for multiple amounts (1.5 ADA, 5 ADA, 10 ADA) using subtests

cardano_node_tests.tests.test_tx_mempool module

Tests for transactions in mempool.

class cardano_node_tests.tests.test_tx_mempool.TestMempool[source]

Bases: object

Tests for transactions in mempool.

payment_addrs_locked(cluster_manager: ClusterManager, cluster_singleton: ClusterLib) → list[AddressRecord][source]: Create 2 new payment addresses for ‘test_query_mempool_txin’.

test_query_mempool_txin(cluster_singleton: ClusterLib, payment_addrs_locked: list[AddressRecord])[source]

Query txin of a transaction that is still in mempool.

Test mempool query commands while transaction is pending in mempool.

Wait for existing transactions to be removed from mempool
Build and sign transaction from source address to destination address with 2 ADA
Submit transaction and re-submit to ensure it stays in mempool
Query UTxO while transaction is in mempool
Check that ‘query tx-mempool next-tx’ returns a TxId
Check that ‘query tx-mempool exists <TxId>’ finds the expected TxId
Verify slot numbers match between mempool query commands

cardano_node_tests.tests.test_tx_metadata module

Tests for transactions with metadata.

class cardano_node_tests.tests.test_tx_metadata.TestMetadata[source]

Bases: object

Tests for transactions with metadata.

CBOR_METADATA_FILE = PosixPath('/home/martink/Source/repos/cardano-node-tests/cardano_node_tests/tests/data/tx_metadata.cbor')[source]

JSON_METADATA_FILE = PosixPath('/home/martink/Source/repos/cardano-node-tests/cardano_node_tests/tests/data/tx_metadata.json')[source]

JSON_METADATA_INVALID_FILE = PosixPath('/home/martink/Source/repos/cardano-node-tests/cardano_node_tests/tests/data/tx_metadata_invalid.json')[source]

JSON_METADATA_LONG_FILE = PosixPath('/home/martink/Source/repos/cardano-node-tests/cardano_node_tests/tests/data/tx_metadata_long.json')[source]

JSON_METADATA_WRONG_FILE = PosixPath('/home/martink/Source/repos/cardano-node-tests/cardano_node_tests/tests/data/tx_metadata_wrong.json')[source]

METADATA_DUPLICATES = 'tx_metadata_duplicate_keys*.json'[source]

payment_addr(cluster_manager: ClusterManager, cluster: ClusterLib) → AddressRecord[source]: Create new payment address.

test_build_tx_invalid_json_metadata(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Try to build transaction with invalid (malformed) metadata JSON.

Expect failure.

Uses cardano-cli transaction build command for building the transactions. The metadata file contains invalid JSON syntax.

Prepare transaction files with invalid JSON metadata file
Attempt to build transaction using build
Check that transaction building fails with JSON parse error

test_build_tx_metadata_both(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with both metadata JSON and CBOR formats combined.

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

Prepare transaction files with both JSON and CBOR metadata files
Build, sign and submit transaction with both metadata types attached
Load metadata from transaction body CBOR
Merge expected metadata from both JSON and CBOR files
Check that combined metadata in TX body matches merged original metadata
Verify transaction view command output matches metadata
(optional) Check transactions and metadata in db-sync

test_build_tx_metadata_cbor(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with metadata CBOR and verify metadata preservation.

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

Prepare transaction files with metadata CBOR file
Build, sign and submit transaction from payment address with CBOR metadata
Load metadata from transaction body CBOR
Compare transaction body metadata with original CBOR file
Check that metadata in TX body matches the original metadata exactly
(optional) Check transactions and metadata in db-sync

test_build_tx_metadata_json(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with metadata JSON and verify metadata preservation.

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

Prepare transaction files with metadata JSON file
Build, sign and submit transaction from payment address with metadata
Load metadata from transaction body CBOR
Compare transaction body metadata with original JSON file
Check that metadata in TX body matches the original metadata exactly
(optional) Check transactions and metadata in db-sync

test_build_tx_too_long_metadata_json(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Try to build transaction with metadata JSON string exceeding 64 bytes.

Expect failure.

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

Prepare transaction files with metadata JSON containing string > 64 UTF-8 bytes
Attempt to build transaction using build
Check that transaction building fails with string length error

test_build_tx_wrong_json_metadata_format(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Try to build transaction with incorrectly formatted metadata JSON.

Expect failure.

Uses cardano-cli transaction build command for building the transactions. The metadata file is valid JSON but not in expected format (map at top level).

Prepare transaction files with wrongly formatted metadata JSON
Attempt to build transaction using build
Check that transaction building fails with metadata format error

test_tx_duplicate_metadata_keys(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with multiple metadata JSON files containing duplicate keys.

Uses cardano-cli transaction build-raw command.

Prepare transaction files with multiple JSON metadata files having duplicate keys
Send transaction from payment address with all metadata files attached
Load metadata from transaction body CBOR
Merge expected metadata from input files (first occurrence wins for duplicates)
Check that metadata in TX body matches merged metadata with correct precedence
Verify that for duplicate keys the first occurrence is used
(optional) Check transactions and metadata in db-sync

test_tx_invalid_json_metadata(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Try to build transaction with invalid (malformed) metadata JSON.

Expect failure.

Uses cardano-cli transaction build-raw command. The metadata file contains invalid JSON syntax.

Prepare transaction files with invalid JSON metadata file
Attempt to build transaction using build-raw
Check that transaction building fails with JSON parse error

test_tx_metadata_both(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with both metadata JSON and CBOR formats combined.

Uses cardano-cli transaction build-raw command.

Prepare transaction files with both JSON and CBOR metadata files
Send transaction from payment address with both metadata types attached
Load metadata from transaction body CBOR
Merge expected metadata from both JSON and CBOR files
Check that combined metadata in TX body matches merged original metadata
Verify transaction view command output matches metadata
(optional) Check transactions and metadata in db-sync

test_tx_metadata_cbor(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with metadata CBOR and verify metadata preservation.

Uses cardano-cli transaction build-raw command.

Prepare transaction files with metadata CBOR file
Send transaction from payment address with CBOR metadata attached
Load metadata from transaction body CBOR
Compare transaction body metadata with original CBOR file
Check that metadata in TX body matches the original metadata exactly
(optional) Check transactions and metadata in db-sync

test_tx_metadata_json(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with metadata JSON and verify metadata preservation.

Uses cardano-cli transaction build-raw command.

Prepare transaction files with metadata JSON file
Send transaction from payment address with metadata attached
Load metadata from transaction body CBOR
Compare transaction body metadata with original JSON file
Check that metadata in TX body matches the original metadata exactly
(optional) Check transactions and metadata in db-sync

test_tx_metadata_no_txout(cluster: ClusterLib, cluster_manager: ClusterManager)[source]

Send transaction with just metadata, no UTxO is produced.

Submit a transaction where all funds available on source address is used for fee
Check that no UTxOs are created by the transaction
Check that there are no funds left on source address
Check that the metadata in TX body matches the original metadata

test_tx_too_long_metadata_json(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Try to build transaction with metadata JSON string exceeding 64 bytes.

Expect failure.

Uses cardano-cli transaction build-raw command.

Prepare transaction files with metadata JSON containing string > 64 UTF-8 bytes
Attempt to build transaction using build-raw
Check that transaction building fails with string length error

test_tx_wrong_json_metadata_format(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Try to build transaction with incorrectly formatted metadata JSON.

Expect failure.

Uses cardano-cli transaction build-raw command. The metadata file is valid JSON but not in expected format (map at top level).

Prepare transaction files with wrongly formatted metadata JSON
Attempt to build transaction using build-raw
Check that transaction building fails with metadata format error

cardano_node_tests.tests.test_tx_negative module

Negative tests for transactions.

Tests like duplicated transaction, sending funds to wrong addresses, wrong fee, wrong ttl.

class cardano_node_tests.tests.test_tx_negative.TestNegative[source]

Bases: object

Transaction tests that are expected to fail.

cluster_wrong_tx_era(skip_on_last_era: None, skip_unknown_last_era: None, cluster: ClusterLib) → ClusterLib[source]

pool_users(cluster_manager: ClusterManager, cluster: ClusterLib) → list[PoolUser][source]: Create pool users.

skip_on_last_era() → None[source]

skip_unknown_last_era() → None[source]

test_before_negative_overflow(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str)[source]

Try to send transaction with negative invalid_before causing integer overflow.

Expect failure.

Uses parametrized build method (build, build-raw, or build-estimate). Negative values overflow to MAX_UINT64, which exceeds valid range.

Set invalid_before to -5 (overflows to MAX_UINT64 - 5)
Attempt to build or submit transaction with overflowed value
Check that transaction fails or builds with clamped value
Verify validity range in transaction view output

test_before_positive_overflow(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str)[source]

Try to send transaction with invalid_before exceeding MAX_UINT64 causing overflow.

Expect failure.

Uses parametrized build method (build, build-raw, or build-estimate). Values > MAX_UINT64 overflow back to valid range.

Set invalid_before to MAX_UINT64 + (MAX_INT64 + 5) which overflows
Attempt to build or submit transaction with overflowed value
Check that transaction fails or builds with clamped value
Verify validity range in transaction view output

test_before_too_high(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str)[source]

Try to send transaction with invalid_before exceeding MAX_INT64.

Expect failure.

Uses parametrized build method (build, build-raw, or build-estimate). Valid values must be <= MAX_INT64.

Set invalid_before to MAX_INT64 + 5 (exceeds valid range)
Attempt to build or submit transaction with too-high value
Check that transaction fails with OutsideValidityIntervalUTxO error

test_build_invalid_length_utxo_hash(cluster: ClusterLib, pool_users: list[PoolUser]) → None[source]

Try to use UTxO hash with invalid length as input (property-based test).

Expect failure.

Uses cardano-cli transaction build command for building the transactions. Uses hypothesis to generate random 10-550 char hashes.

Generate random UTxO hash string with incorrect length
Attempt to build transaction using wrong-length UTxO hash
Check that transaction building fails with deserialization or format error

test_build_missing_change_address(cluster: ClusterLib, pool_users: list[PoolUser])[source]

Try to build transaction with missing –change-address parameter.

Expect failure.

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

Prepare transaction inputs and outputs
Attempt to build transaction using build without –change-address parameter
Check that transaction building fails with missing change address error

test_build_missing_tx_in(cluster: ClusterLib, pool_users: list[PoolUser])[source]

Try to build transaction with missing –tx-in parameter.

Expect failure.

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

Prepare transaction outputs only (no inputs)
Attempt to build transaction using build without –tx-in parameter
Check that transaction building fails with missing tx-in error

test_build_multiple_change_addresses(cluster: ClusterLib, pool_users: list[PoolUser])[source]

Try to build transaction with multiple –change-address parameters.

Expect failure.

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

Prepare transaction inputs and outputs
Attempt to build transaction with two different –change-address parameters
Check that transaction building fails with invalid option error

test_build_send_funds_invalid_change_address(cluster: ClusterLib, pool_users: list[PoolUser]) → None[source]

Try to send funds using invalid change address (property-based test).

Expect failure.

Uses cardano-cli transaction build command for building the transactions. Uses hypothesis to generate random 98-char addresses.

Generate random address string with valid characters but invalid format
Attempt to build transaction with invalid change address
Check that transaction building fails with invalid address error

test_build_send_funds_invalid_chars_change_address(cluster: ClusterLib, pool_users: list[PoolUser]) → None[source]

Try to send funds using change address with invalid characters (property-based test).

Expect failure.

Uses cardano-cli transaction build command for building the transactions. Uses hypothesis to generate random 98-char strings.

Generate random address string with invalid characters
Attempt to build transaction with change address containing bad characters
Check that transaction building fails with invalid address error

test_build_send_funds_invalid_length_change_address(cluster: ClusterLib, pool_users: list[PoolUser]) → None[source]

Try to send funds using change address with invalid length (property-based test).

Expect failure.

Uses cardano-cli transaction build command for building the transactions. Uses hypothesis to generate random 50-250 char addresses.

Generate random address string with incorrect length
Attempt to build transaction with wrong-length change address
Check that transaction building fails with invalid address error

test_duplicated_tx(cluster: ClusterLib, pool_users: list[PoolUser])[source]

Try to submit an identical transaction twice.

Expect failure.

Uses cardano-cli transaction build-raw command.

Build and sign transaction sending 2 ADA from source to destination address
Submit transaction successfully the first time
Check expected balances for both source and destination addresses
Attempt to submit the exact same signed transaction again
Check that re-submission fails with inputs already spent error

test_invalid_length_utxo_hash(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str) → None[source]

Try to use UTxO hash with invalid length as input (property-based test).

Expect failure.

Uses parametrized build method and hypothesis to generate random 10-550 char hashes.

Generate random UTxO hash string with incorrect length
Attempt to build or submit transaction using wrong-length UTxO hash
Check that transaction fails with deserialization or format error

test_lower_bound_not_supported(cluster: ClusterLib, pool_users: list[PoolUser])[source]

Try to build Shelley era transaction with unsupported –invalid-before.

Expect failure.

Uses cardano-cli transaction build-raw command. Only runs with Shelley TX era (validity lower bound not supported).

Prepare transaction from source to destination with 2 ADA
Attempt to build transaction with invalid_before=10 (unsupported in Shelley)
Check that transaction building fails with lower bound not supported error

test_missing_fee(cluster: ClusterLib, pool_users: list[PoolUser])[source]

Try to build transaction with missing –fee parameter.

Expect failure.

Uses cardano-cli transaction build-raw command.

Prepare transaction inputs and outputs
Attempt to build transaction using build-raw without –fee parameter
Check that transaction building fails with missing fee error

test_missing_ttl(cluster: ClusterLib, pool_users: list[PoolUser])[source]

Try to build Shelley era transaction with missing –ttl parameter.

Expect failure.

Uses cardano-cli transaction build-raw command. Only runs with Shelley TX era on node < 8.7.0.

Prepare transaction inputs and outputs
Attempt to build Shelley transaction without –invalid-hereafter parameter
Check that transaction building fails with missing TTL error

test_missing_tx_in(cluster: ClusterLib, pool_users: list[PoolUser])[source]

Try to build transaction with missing –tx-in parameter.

Expect failure.

Uses cardano-cli transaction build-raw command.

Prepare transaction outputs only (no inputs)
Attempt to build transaction using build-raw without –tx-in parameter
Check that transaction building fails with missing tx-in error

test_nonexistent_utxo_hash(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str)[source]

Try to use nonexistent UTxO hash as transaction input.

Expect failure.

Uses parametrized build method (build or build-raw, not build-estimate).

Get valid UTxO from payment address
Modify last 4 characters of UTxO hash to create nonexistent hash
Attempt to build or submit transaction using invalid UTxO hash
Check that transaction fails with empty UTxO or BadInputsUTxO error

test_nonexistent_utxo_ix(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str)[source]

Try to use nonexistent UTxO transaction index (TxIx) as input.

Expect failure.

Uses parametrized build method (build or build-raw, not build-estimate).

Get valid UTxO from payment address
Modify UTxO index to nonexistent value (5)
Attempt to build or submit transaction using invalid UTxO index
Check that transaction fails with empty UTxO or BadInputsUTxO error

test_past_ttl(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str)[source]

Try to send transaction with TTL (time-to-live) in the past.

Expect failure.

Uses parametrized build method (build, build-raw, or build-estimate).

Set invalid_hereafter to current slot - 1 (already expired)
Attempt to build or submit transaction with past TTL
Check that transaction fails with ExpiredUTxO or OutsideValidityIntervalUTxO error

test_pbt_before_negative_overflow(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str) → None[source]

Try to send transaction with negative invalid_before (property-based test).

Expect failure.

Uses parametrized build method and hypothesis with values 1 to MAX_INT64. Negative values cause integer overflow.

Set invalid_before to negative of parametrized value
Attempt to build or submit transaction with negative value
Check that transaction fails or slot number is positive after overflow

test_pbt_before_positive_overflow(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str) → None[source]

Try to send transaction with invalid_before > MAX_UINT64 (property-based test).

Expect failure.

Uses parametrized build method and hypothesis with values (MAX_INT64+1) to MAX_UINT64. Values > MAX_UINT64 cause integer overflow.

Set invalid_before to MAX_UINT64 + parametrized value (causes overflow)
Attempt to build or submit transaction with overflowed value
Check that transaction fails or verifies overflow behavior

test_pbt_before_too_high(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str) → None[source]

Try to send transaction with invalid_before > MAX_INT64 (property-based test).

Expect failure.

Uses parametrized build method and hypothesis with values (MAX_INT64+1) to MAX_UINT64. Valid values must be <= MAX_INT64.

Set invalid_before to parametrized value exceeding MAX_INT64
Attempt to build or submit transaction with too-high value
Check that transaction fails with OutsideValidityIntervalUTxO error
Verify slot number matches expected value

test_send_funds_from_invalid_address(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str) → None[source]

Try to send funds from randomly generated invalid address (property-based test).

Expect failure.

Uses parametrized build method and hypothesis to generate random 98-char addresses.

Generate random address string with valid characters but invalid format
Attempt to build transaction from invalid source address to valid destination
Check that transaction building fails with invalid address error

test_send_funds_from_invalid_chars_address(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str) → None[source]

Try to send funds from address with invalid characters (property-based test).

Expect failure.

Uses parametrized build method and hypothesis to generate random 98-char strings.

Generate random address string with invalid characters
Attempt to build transaction from address with bad characters to valid destination
Check that transaction building fails with invalid address error

test_send_funds_from_invalid_length_address(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str) → None[source]

Try to send funds from address with invalid length (property-based test).

Expect failure.

Uses parametrized build method and hypothesis to generate random 50-250 char addresses.

Generate random address string with incorrect length
Attempt to build transaction from wrong-length source address to valid destination
Check that transaction building fails with invalid address error

test_send_funds_to_invalid_address(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str) → None[source]

Try to send funds to randomly generated invalid address (property-based test).

Expect failure.

Uses parametrized build method and hypothesis to generate random 98-char addresses.

Generate random address string with valid characters but invalid format
Attempt to build transaction with 1 ADA to invalid address
Check that transaction building fails with invalid address error

test_send_funds_to_invalid_chars_address(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str) → None[source]

Try to send funds to address with invalid characters (property-based test).

Expect failure.

Uses parametrized build method and hypothesis to generate random 98-char strings.

Generate random address string with invalid characters
Attempt to build transaction with 1 ADA to address with bad characters
Check that transaction building fails with invalid address error

test_send_funds_to_invalid_length_address(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str) → None[source]

Try to send funds to address with invalid length (property-based test).

Expect failure.

Uses parametrized build method and hypothesis to generate random 50-250 char addresses.

Generate random address string with incorrect length
Attempt to build transaction with 1 ADA to wrong-length address
Check that transaction building fails with invalid address error

test_send_funds_to_reward_address(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str)[source]

Try to send funds from payment address to stake (reward) address.

Expect failure.

Uses parametrized build method (build, build-raw, or build-estimate).

Prepare transaction with stake address as destination (invalid)
Attempt to build transaction with 1 ADA to stake address
Check that transaction building fails with invalid address error

test_send_funds_to_utxo_address(cluster: ClusterLib, pool_users: list[PoolUser], build_method: str)[source]

Try to send funds from payment address to UTxO hash (not an address).

Expect failure.

Uses parametrized build method (build, build-raw, or build-estimate).

Get UTxO hash from payment address query
Prepare transaction with UTxO hash as destination (invalid)
Attempt to build transaction with 1 ADA to UTxO hash
Check that transaction building fails with invalid address error

test_wrong_network_magic(cluster: ClusterLib, pool_users: list[PoolUser])[source]

Try to submit transaction with incorrect network magic number.

Expect failure.

Uses cardano-cli transaction build-raw command.

Build and sign transaction sending 2 ADA from source to destination address
Add ignore rules for expected handshake errors in logs
Attempt to submit transaction with network magic + 100
Check that submission fails with HandshakeError

test_wrong_signing_key(cluster: ClusterLib, pool_users: list[PoolUser])[source]

Try to send transaction signed with incorrect signing key.

Expect failure.

Uses cardano-cli transaction send command.

Prepare transaction from pool_users[0] to pool_users[1] with 1.5 ADA
Sign transaction with pool_users[1] signing key instead of pool_users[0]
Attempt to build and submit transaction with wrong signing key
Check that transaction fails with MissingVKeyWitnessesUTXOW error

test_wrong_tx_era(cluster: ClusterLib, cluster_wrong_tx_era: ClusterLib, pool_users: list[PoolUser])[source]

Try to send transaction using TX era higher than network era.

Expect failure.

Uses cardano-cli transaction send command with future era.

Prepare transaction from pool_users[0] to pool_users[1] with 1.5 ADA
Use cluster instance configured for era > current network era
Attempt to submit transaction built for future era
Check that transaction fails with era mismatch error

cardano_node_tests.tests.test_tx_unbalanced module

Tests for unbalanced transactions.

class cardano_node_tests.tests.test_tx_unbalanced.TestUnbalanced[source]

Bases: object

Tests for unbalanced transactions.

payment_addrs(cluster_manager: ClusterManager, cluster: ClusterLib) → list[AddressRecord][source]: Create 2 new payment addresses.

pbt_highest_utxo(cluster: ClusterLib, payment_addrs: list[AddressRecord]) → UTXOData[source]

Get UTxO with highest amount of Lovelace.

Meant for property-based tests, so this expensive operation gets executed only once.

test_build_transfer_amount_bellow_minimum(cluster: ClusterLib, payment_addrs: list[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Try to build transaction with amount below minimum UTxO threshold (property-based test).

Expect failure.

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

Use UTxO with highest amount as sole input
Attempt to create transaction output with amount below minimum UTxO (0 to ~1 ADA)
Check that transaction building fails with minimum UTxO threshold error

test_build_transfer_negative_amount(cluster: ClusterLib, payment_addrs: list[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Try to build transaction with negative Lovelace amount (property-based test).

Expect failure.

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

Use UTxO with highest amount as sole input
Attempt to create transaction output with negative amount (-MAX to -1)
Check that transaction building fails with negative quantity error

test_build_transfer_unavailable_funds(cluster: ClusterLib, payment_addrs: list[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Try to build transaction transferring more funds than available (property-based test).

Expect failure.

Uses cardano-cli transaction build command.

Use UTxO with highest amount as sole input
Attempt to transfer amount exceeding UTxO balance by parametrized value (1 to MAX/2)
Check that transaction building fails with negative balance error

test_negative_change(cluster: ClusterLib, payment_addrs: list[AddressRecord])[source]

Try to build a transaction with a negative change amount.

Expect failure.

Uses cardano-cli transaction build-raw command.

Calculate transaction fee
Get UTxO with highest amount from source address
Attempt to create transaction outputs that exceed inputs by 1 Lovelace
Include a change output with -1 Lovelace
Check that transaction building fails with appropriate error

test_out_of_bounds_amount(cluster: ClusterLib, payment_addrs: list[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Try to build transaction with Lovelace amount exceeding maximum (property-based test).

Expect failure.

Uses cardano-cli transaction build-raw command.

Use UTxO with highest amount as sole input
Attempt to create transaction output with amount > MAX_UINT64
Check that transaction building fails with out of bounds error

test_transfer_amount_bellow_minimum(cluster: ClusterLib, payment_addrs: list[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Try to build and submit transaction with amount below minimum UTxO (property-based test).

Expect failure.

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

Use UTxO with highest amount as sole input
Build transaction with output amount below minimum UTxO (0 to ~1 ADA)
Sign the transaction successfully (no validation at build-raw stage)
Attempt to submit the transaction
Check that submission fails with OutputTooSmallUTxO error

test_transfer_negative_amount(cluster: ClusterLib, payment_addrs: list[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Try to build transaction with negative Lovelace amount (property-based test).

Expect failure.

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

Use UTxO with highest amount as sole input
Attempt to build transaction with negative output amount (-MAX to -1)
Check that transaction building fails with negative quantity error

test_wrong_balance(cluster: ClusterLib, payment_addrs: list[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Build transaction with unbalanced change amount (property-based test).

Expect failure.

Uses cardano-cli transaction build-raw command.

Use UTxO with highest amount as sole input
Transfer full balance minus fee to destination address
Add incorrect change output with parametrized amount (2 ADA to MAX)
Build and sign the unbalanced transaction successfully
Attempt to submit the unbalanced transaction
Check that submission fails with ValueNotConservedUTxO error

cardano_node_tests.tests.test_xdist_helper module

Dummy test that helps with pytest-xdist scheduling.

The pytest-xdist plugin needs to schedule two tests per worker in initial batch. This dummy test gets scheduled as first not long-running test on every pytest worker so the other test can be an actual long-running test.

cardano_node_tests.tests.test_xdist_helper.test_dummy(for_worker: int) → None[source]

cardano_node_tests.tests.tx_common module

cardano_node_tests.tests.tx_common.get_raw_tx_values(cluster_obj: ClusterLib, tx_name: str, src_record: AddressRecord, dst_record: AddressRecord, for_build_command: bool = False) → TxRawOutput[source]: Get values for manually building Tx using transaction build or transaction build-raw.

cardano_node_tests.tests.tx_common.get_txins_txouts(txins: list[UTXOData], txouts: list[TxOut]) → tuple[list[str], list[str]][source]

cardano_node_tests.tests package

Subpackages

Submodules

cardano_node_tests.tests.common module

cardano_node_tests.tests.conftest module

cardano_node_tests.tests.delegation module

cardano_node_tests.tests.issues module

cardano_node_tests.tests.kes module

cardano_node_tests.tests.plutus_common module

cardano_node_tests.tests.reqs_conway module

cardano_node_tests.tests.test_addr_registration module

cardano_node_tests.tests.test_blocks module

cardano_node_tests.tests.test_chain_transactions module

cardano_node_tests.tests.test_cli module

cardano_node_tests.tests.test_configuration module

cardano_node_tests.tests.test_dbsync module

cardano_node_tests.tests.test_dbsync_config module

cardano_node_tests.tests.test_delegation module

cardano_node_tests.tests.test_env_network_id module

cardano_node_tests.tests.test_kes module

cardano_node_tests.tests.test_ledger_state module

cardano_node_tests.tests.test_metrics module

cardano_node_tests.tests.test_mir_certs module

cardano_node_tests.tests.test_mnemonic module

cardano_node_tests.tests.test_native_tokens module

cardano_node_tests.tests.test_node_upgrade module

cardano_node_tests.tests.test_pool_saturation module

cardano_node_tests.tests.test_pools module

cardano_node_tests.tests.test_protocol module

cardano_node_tests.tests.test_reconnect module

cardano_node_tests.tests.test_rollback module

cardano_node_tests.tests.test_scripts module

cardano_node_tests.tests.test_smash module

cardano_node_tests.tests.test_socket_path module

cardano_node_tests.tests.test_staking_no_rewards module

cardano_node_tests.tests.test_staking_rewards module

cardano_node_tests.tests.test_tx_basic module

cardano_node_tests.tests.test_tx_fees module

cardano_node_tests.tests.test_tx_many_utxos module

cardano_node_tests.tests.test_tx_mempool module

cardano_node_tests.tests.test_tx_metadata module

cardano_node_tests.tests.test_tx_negative module

cardano_node_tests.tests.test_tx_unbalanced module

cardano_node_tests.tests.test_xdist_helper module

cardano_node_tests.tests.tx_common module

Module contents