cardano_node_tests.tests package

Subpackages

Submodules

cardano_node_tests.tests.common module

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.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_test_id(cluster_obj: ClusterLib) → 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.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.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.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_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, use_build_cmd: bool = False) → DelegationOut[source]: Submit registration certificate and delegate to 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.Path, datum_file: pathlib.Path | None = None, datum_cbor_file: pathlib.Path | None = None, datum_value: str | None = None, redeemer_file: pathlib.Path | None = None, redeemer_cbor_file: pathlib.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.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]

class cardano_node_tests.tests.plutus_common.Token(coin: str, amount: int)[source]

Bases: object

amount: int[source]

coin: str[source]

cardano_node_tests.tests.plutus_common.check_plutus_costs(plutus_costs: List[dict], expected_costs: List[ExecutionCost], frac: float = 0.15)[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], use_build_cmd: bool)[source]: Deregister not registered stake address.

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.

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.

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], use_build_cmd: bool, submit_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], use_build_cmd: bool)[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], use_build_cmd: bool)[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

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.

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

pytestmark = [Mark(name='skipif', args=(False,), kwargs={'reason': 'runs only on P2P enabled clusters'})][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.

skip_leadership_schedule()[source]

test_pool_blocks(skip_leadership_schedule: None, cluster_manager: ClusterManager, cluster_use_pool: Tuple[ClusterLib, str], for_epoch: str)[source]

Check that blocks were minted 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 minted blocks in queried epoch for the selected pool
compare leadership schedule with blocks that were actually minted
compare db-sync records with ledger state dump

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

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

Expect failure.

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 transactions chaining.

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 doesn’t exist on ledger yet. Node needs to lookup the Tx in mempool and correctly chain the dependent transactons.

cardano_node_tests.tests.test_chain_transactions.get_payment_addr(cluster_manager: ClusterManager, cluster_obj: ClusterLib) → AddressRecord[source]: Create new payment address.

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]: Check address build with all valid input options.

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]: Check payment address info.

test_address_info_payment_with_outfile(cluster: ClusterLib)[source]: Compare payment address info with and without outfile provided.

test_address_info_script(cluster: ClusterLib)[source]: Check script address info.

test_address_info_stake(cluster: ClusterLib, addr_gen: str)[source]: Check stake address info.

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]: Check address key-hash with valid verification key.

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]: Test query ledger-state.

test_pool_params(cluster: ClusterLib, pool_ids: List[str])[source]: Test query pool-params.

test_pool_state(cluster: ClusterLib, pool_ids: List[str])[source]: Test query pool-state.

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

Test query stake-snapshot.

See also TestLedgerState.test_stake_snapshot for more scenarios.

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_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]: Check the calculate-min-fee command.

test_protocol_mode(cluster: ClusterLib)[source]: Check the default protocol mode - command works even without specifying protocol mode.

test_query_tip(cluster: ClusterLib)[source]: Test query tip.

test_sign_tx_with_process_substitution(cluster: ClusterLib)[source]: Check that it is possible to pass skey file using process substitution.

test_toplevel_queries(cluster: ClusterLib, command: str)[source]: Check that various queries are available in top level.

test_tx_view(cluster: ClusterLib)[source]: Check that the output of transaction view is as expected.

test_txid_with_process_substitution(cluster: ClusterLib)[source]: Check that it is possible to pass Tx file using process substitution.

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]: Check that the non-extended verification key is according the verification key.

test_stake_non_extended_key(cluster: ClusterLib)[source]: Get a stake non-extended-key from a stake extended key.

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

Bases: object

Tests for cardano-cli ping.

test_ping_localhost(cluster: ClusterLib)[source]: Test cardano-cli ping on local node using host and port.

test_ping_unix_socket(cluster: ClusterLib)[source]: Test cardano-cli ping on local node using unix socket.

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

Bases: object

Tests for cardano-cli query slot-number.

test_slot_number(cluster: ClusterLib)[source]: Test query slot-number.

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]: Check that pretty printed query utxo output looks as expected.

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]: Check that it is possible to return the whole UTxO on local cluster.

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]: Check stake-address key-hash with valid verification key.

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_slot_length(cluster_slot_length: ClusterLib)[source]: Test the slotLength configuration.

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_cost_model(cluster: ClusterLib)[source]: Check expected values in the cost_model table in db-sync.

test_epoch(cluster: ClusterLib)[source]: Check expected values in the epoch table in db-sync.

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.

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, use_build_cmd: bool)[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_using_pool_id(cluster_manager: ClusterManager, cluster_and_pool: Tuple[ClusterLib, str], use_build_cmd: bool)[source]

Submit registration certificate and delegate to pool using pool id.

register stake address and delegate it to pool
check that the stake address was delegated
(optional) check records in db-sync

test_delegate_using_vkey(cluster_manager: ClusterManager, cluster_use_pool: Tuple[ClusterLib, str], use_build_cmd: bool)[source]

Submit registration certificate and delegate to pool using cold vkey.

register stake address and delegate it to pool
check that the stake address was delegated
(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.

Create 1 payment vkey/skey key pair
Create 4 stake vkey/skey key pairs
Create 1 payment addresses for each combination of payment_vkey/stake_vkey
Delegate the stake addresses to 2 different pools
Check that the stake addresses received rewards

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.

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], use_build_cmd: bool)[source]

Try to delegate deregistered stake address.

Expect failure.

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], use_build_cmd: bool)[source]

Try to delegate unknown stake address.

Expect failure.

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.

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.

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]

Send funds to payment address.

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

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]

Test query pool-state.

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]

Test query protocol-parameters.

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]

Test query protocol-state.

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]

Test query stake-address-info.

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]

Test query stake-distribution.

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]: Test query pool-state.

test_query_protocol_params(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib)[source]: Test query protocol-parameters.

test_query_protocol_state(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib)[source]: Test query protocol-state.

test_query_stake_addr_info(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib)[source]: Test query stake-address-info.

test_query_stake_distribution(skip_on_no_env: None, set_network_id_env: None, cluster: ClusterLib)[source]: Test query stake-distribution.

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_governance module

Tests for governance functionality.

Tests for update proposals are in separate file test_update_proposals.py.

This file tests: * poll creation * poll answer * poll verification

class cardano_node_tests.tests.test_governance.TestPoll[source]

Bases: object

Tests for SPO poll.

governance_poll_available() → None[source]

payment_addr(cluster_manager: ClusterManager, cluster: ClusterLib) → AddressRecord[source]: Create new payment address.

test_answer_golden_poll(cluster: ClusterLib, payment_addr: AddressRecord, governance_poll_available: None, tx_file_type: str, use_build_cmd: bool)[source]

Test answering a golden SPO poll.

create an answer to a poll
create Tx with the answer
verify poll answer

test_create_and_answer_poll(cluster_manager: ClusterManager, cluster: ClusterLib, payment_addr: AddressRecord, governance_poll_available: None, use_build_cmd: bool, required_signer_option: str)[source]

Test creating and answering new SPO poll.

create the poll
publish the poll on chain
check that the created Tx has the expected metadata
answer the poll
publish the answer on chain
verify poll answer
(optional) check create-poll transaction in db-sync

test_create_answer_negative_index(cluster: ClusterLib, governance_poll_available: None)[source]

Test answering an SPO poll with an answer with a negative index.

Expect failure.

test_create_invalid_answer(cluster: ClusterLib, governance_poll_available: None) → None[source]

Test answering an SPO poll with invalid answer.

Expect failure.

test_verify_answer_without_required_signer(cluster: ClusterLib, payment_addr: AddressRecord, governance_poll_available: None, tx_file_type: str, use_build_cmd: bool)[source]

Test verifying an answer to an SPO poll without valid required signer.

Expect failure.

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 minting blocks
wait for KES expiration on the selected pool
check that the pool with expired KES didn’t mint 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.

Expect failure.

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

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

Expect failure.

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 minting 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 minting any blocks
generate new operational certificate with valid –kes-period and restart the node
check that the pool is minting 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 minting new blocks
check kes-period-info while the pool is not minting blocks
start the node with the new operational certificate
check that the pool is minting blocks again
check that metrics reported by kes-period-info got updated once the pool started minting 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.

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.

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 metrics == list of expected 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.

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

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.

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

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.

Expected to fail when Era < Alonzo.

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.

Expected to fail when Era < Alonzo.

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_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], use_build_cmd: bool, 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], use_build_cmd: bool, 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, use_build_cmd: bool, 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, use_build_cmd: bool, 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], use_build_cmd: bool, 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], use_build_cmd: bool, 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], use_build_cmd: bool, 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], use_build_cmd: bool, 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], use_build_cmd: bool, 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.

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 that it’s NOT possible to mint tokens when the policy is not met.

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

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

Test that it’s NOT possible to mint tokens when the policy is not met.

The “after” slot is in the past.

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

Test that it’s NOT possible to mint tokens when the policy is not met.

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

test_policy_before_past(cluster: ClusterLib, issuers_addrs: List[AddressRecord])[source]: Test that it’s NOT possible to mint tokens when the “before” slot is in the past.

test_valid_policy_after(cluster: ClusterLib, issuers_addrs: List[AddressRecord], use_build_cmd: bool)[source]: Test minting and burning of tokens after a given slot, check fees in Lovelace.

test_valid_policy_before(cluster: ClusterLib, issuers_addrs: List[AddressRecord], use_build_cmd: bool)[source]: Test minting and burning of tokens before a given slot, check fees in Lovelace.

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], use_build_cmd: bool, 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]) → TokenRecord[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: TokenRecord, use_build_cmd: bool) → None[source]: Test sending an invalid amount of tokens to payment address.

test_transfer_multiple_tokens(cluster: ClusterLib, payment_addrs: List[AddressRecord], new_token: TokenRecord, use_build_cmd: bool)[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: TokenRecord, use_build_cmd: bool)[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: TokenRecord, amount: int, use_build_cmd: bool)[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_ignore_log_errors(cluster_singleton: ClusterLib, worker_id: str)[source]: Ignore selected errors in log right after node upgrade.

test_update_cost_models(cluster_manager: ClusterManager, cluster_singleton: ClusterLib, pool_user_singleton: PoolUser)[source]: Test cost model update.

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], use_build_cmd: bool, 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.

test_submit_tx(cluster_manager: ClusterManager, cluster: ClusterLib, use_build_cmd: bool, for_step: int, from_step: int, file_type: str)[source]: Submit transaction that was created by previous node version and/or in previous era.

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_minted: Dict[int, int], saturation_amounts: Dict[int, int])[source]

Bases: object

blocks_minted: 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, use_build_cmd: bool)[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.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, use_build_cmd: bool)[source]

Create and register a stake pool (without metadata).

Check that pool was registered.

test_deregister_stake_pool(cluster_manager: ClusterManager, cluster: ClusterLib, use_build_cmd: bool)[source]

Deregister stake pool.

deregister stake pool
check that the stake addresses are no longer delegated
check that the pool deposit was returned to reward account

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.

deregister stake pool
check that the stake addresses are no longer delegated
reregister the pool by resubmitting the pool registration certificate
delegate stake address to pool again (the address is already registered)
check that pool was correctly setup
check that the stake addresses were delegated

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, use_build_cmd: bool)[source]

Create and register a stake pool with metadata.

Check that pool was registered and stake address delegated.

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

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

Check that pool was registered and stake address delegated.

test_update_stake_pool_metadata(cluster_manager: ClusterManager, cluster: ClusterLib, use_build_cmd: bool, 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, use_build_cmd: bool, 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_protocol_state_keys(cluster: ClusterLib)[source]: Check output of query protocol-state.

test_protocol_state_outfile(cluster: ClusterLib)[source]: Check output file produced by query protocol-state.

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': 'Runs only on local cluster'})][source]

test_metrics_reconnect(cluster_manager: ClusterManager, cluster_singleton: ClusterLib)[source]: Test using metrics that node reconnects after it was restarted.

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=(False,), kwargs={'reason': 'Works only when all nodes have the same topology type'}), 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': 'runs only on local cluster'})][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], use_build_cmd: bool)[source]

Build transaction with invalid auxiliary script.

Expect failure.

test_tx_script_metadata_cbor(cluster: ClusterLib, payment_addrs: List[AddressRecord], use_build_cmd: bool, 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], use_build_cmd: bool, 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], use_build_cmd: bool, 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], use_build_cmd: bool, submit_method: str)[source]: Send funds to and from script address using the all script.

test_multisig_any(cluster: ClusterLib, payment_addrs: List[AddressRecord], use_build_cmd: bool, submit_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], use_build_cmd: bool, submit_method: str)[source]: Send funds to and from script address using the atLeast script.

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], use_build_cmd: bool, submit_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], use_build_cmd: bool, submit_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]

Check that script address length is the same as length of other addresses.

There was an issue that script address was 32 bytes instead of 28 bytes.

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], use_build_cmd: bool, submit_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], use_build_cmd: bool, 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, use_build_cmd: bool, submit_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], use_build_cmd: bool, submit_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, use_build_cmd: bool, submit_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], use_build_cmd: bool, submit_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], use_build_cmd: bool, submit_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.

fund_script_after_slot_in_future(cluster: ClusterLib, payment_addrs: List[AddressRecord], request: SubRequest) → Tuple[Path, str, 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, 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, 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, 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, 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, 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, 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, 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_build_cmd: bool, use_tx_validity: bool)[source]: Check that it is possible to spend from script address after given slot.

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

test_tx_missing_validity(cluster: ClusterLib, payment_addrs: List[AddressRecord], use_build_cmd: bool, 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], use_build_cmd: bool)[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, src_address: str, dst_address: str, amount: int, payment_skey_files: List[Path], multisig_script: Path | None = None, invalid_hereafter: int | None = None, invalid_before: int | None = None, use_build_cmd: bool = False, submit_method: str = 'cli') → TxRawOutput[source]: Build and submit multisig transaction.

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]

Send funds to payment address.

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

Expect failure.

test_neg_query_pool_state(set_socket_path: None, cluster: ClusterLib, env_scenario: str, socket_scenario: str)[source]

Test query pool-state.

Expect failure.

test_neg_query_protocol_params(set_socket_path: None, cluster: ClusterLib, env_scenario: str, socket_scenario: str)[source]

Test query protocol-parameters.

Expect failure.

test_neg_query_protocol_state(set_socket_path: None, cluster: ClusterLib, env_scenario: str, socket_scenario: str)[source]

Test query protocol-state.

Expect failure.

test_neg_query_stake_addr_info(set_socket_path: None, cluster: ClusterLib, env_scenario: str, socket_scenario: str)[source]

Test query stake-address-info.

Expect failure.

test_neg_query_stake_distribution(set_socket_path: None, cluster: ClusterLib, env_scenario: str, socket_scenario: str)[source]

Test query stake-distribution.

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]: Test query pool-state.

test_query_protocol_params(set_socket_path: None, cluster: ClusterLib, env_scenario: str)[source]: Test query protocol-parameters.

test_query_protocol_state(set_socket_path: None, cluster: ClusterLib, env_scenario: str)[source]: Test query protocol-state.

test_query_stake_addr_info(set_socket_path: None, cluster: ClusterLib, env_scenario: str)[source]: Test query stake-address-info.

test_query_stake_distribution(set_socket_path: None, cluster: ClusterLib, env_scenario: str)[source]: Test query stake-distribution.

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 sends all available funds, and extract fee amount from the error message
send 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], use_build_cmd: bool, submit_method: str)[source]

Test default Tx era.

check that default Tx era is implicit
check that default Tx era can be specified explicitly

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.

test_extra_signing_keys(cluster: ClusterLib, payment_addrs: List[AddressRecord], submit_method: str)[source]

Send a transaction with extra signing key.

Check that it is possible to use unneeded signing key in addition to the necessary signing keys for signing the transaction.

test_far_future_ttl(cluster: ClusterLib, payment_addrs: List[AddressRecord], submit_method: str)[source]: Send a transaction with ttl far in the future.

test_funds_to_valid_address(cluster: ClusterLib, payment_addrs: List[AddressRecord], use_build_cmd: bool, submit_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], use_build_cmd: bool, submit_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.

test_missing_tx_out(cluster: ClusterLib, payment_addrs: List[AddressRecord])[source]: Build a transaction with a missing –tx-out parameter.

test_multiple_same_txins(cluster: ClusterLib, payment_addrs: List[AddressRecord], submit_method: str)[source]: Try to build a transaction with multiple identical txins.

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_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, use_build_cmd: bool, submit_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_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], use_build_cmd: bool, 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, use_build_cmd: bool, submit_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, use_build_cmd: bool, submit_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, use_build_cmd: bool, submit_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, use_build_cmd: bool, submit_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.

test_addr_registration_fees(cluster: ClusterLib, pool_users: List[PoolUser], addr_fee: Tuple[int, int])[source]: Test stake address registration fees.

test_pool_deregistration_fees(cluster: ClusterLib, pool_users: List[PoolUser], addr_fee: Tuple[int, int])[source]: Test pool deregistration fees.

test_pool_registration_fees(cluster: ClusterLib, pool_users: List[PoolUser], addr_fee: Tuple[int, int])[source]: Test pool registration fees.

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

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_negative_fee(cluster: ClusterLib, payment_addrs: List[AddressRecord]) → None[source]

Try to send a transaction with negative fee (property-based test).

Expect failure.

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.

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 transaction with many UTxOs (300+) with small amounts of ADA (1-10).

use source address with many UTxOs (100000+)
use destination address with many UTxOs (100000+)
sent transaction with many UTxOs (300+) with tiny amounts of Lovelace from source address to destination address
check expected balances for both source and destination addresses

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]

Test that is possible to query txin of a transaction that is still in mempool.

check if ‘query tx-mempool next-tx’ is returning a TxId
check if ‘query tx-mempool exists <TxId>’ found the expected TxId

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 a transaction with invalid metadata JSON.

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

The metadata file is an invalid JSON.

test_build_tx_metadata_both(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with both metadata JSON and CBOR.

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

Check that the metadata in TX body matches the original metadata.

test_build_tx_metadata_cbor(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with metadata CBOR.

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

Check that the metadata in TX body matches the original metadata.

test_build_tx_metadata_json(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with metadata JSON.

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

check that the metadata in TX body matches the original metadata
(optional) check transactions in db-sync

test_build_tx_too_long_metadata_json(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Try to build a transaction with metadata JSON longer than 64 bytes.

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

test_build_tx_wrong_json_metadata_format(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Try to build a transaction with wrong format of metadata JSON.

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

The metadata file is a valid JSON, but not in a format that is expected.

test_tx_duplicate_metadata_keys(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with multiple metadata JSON files and with duplicate keys.

check that the metadata in TX body matches the original metadata
check that in case of duplicate keys the first occurrence is used

test_tx_invalid_json_metadata(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Try to build a transaction with invalid metadata JSON.

The metadata file is an invalid JSON.

test_tx_metadata_both(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with both metadata JSON and CBOR.

Check that the metadata in TX body matches the original metadata.

test_tx_metadata_cbor(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with metadata CBOR.

Check that the metadata in TX body matches the original metadata.

test_tx_metadata_json(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Send transaction with metadata JSON.

check that the metadata in TX body matches the original metadata
(optional) check transactions 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 a transaction with metadata JSON longer than 64 bytes.

test_tx_wrong_json_metadata_format(cluster: ClusterLib, payment_addr: AddressRecord)[source]

Try to build a transaction with wrong format of metadata JSON.

The metadata file is a valid JSON, but not in a format that is expected.

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], use_build_cmd: bool)[source]

Try to send a transaction with negative invalid_before and check for int overflow.

Expect failure.

test_before_positive_overflow(cluster: ClusterLib, pool_users: List[PoolUser], use_build_cmd: bool)[source]

Try to send a transaction with invalid_before > MAX_UINT64.

Check for int overflow. Expect failure.

test_before_too_high(cluster: ClusterLib, pool_users: List[PoolUser], use_build_cmd: bool)[source]

Try to send a transaction with invalid_before > MAX_INT64.

Expect failure.

test_build_invalid_length_utxo_hash(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to use invalid UTxO hash as an input (property-based test).

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

Expect failure.

test_build_missing_change_address(cluster: ClusterLib, pool_users: List[PoolUser])[source]

Try to build a transaction with a missing –change-address parameter.

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

Expect failure.

test_build_missing_tx_in(cluster: ClusterLib, pool_users: List[PoolUser])[source]

Try to build a transaction with a missing –tx-in parameter.

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

Expect failure.

test_build_multiple_change_addresses(cluster: ClusterLib, pool_users: List[PoolUser])[source]

Try to build a transaction with multiple –change-address parameters.

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

Expect failure.

test_build_send_funds_from_invalid_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from non-existent address (property-based test).

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

Expect failure.

test_build_send_funds_from_invalid_chars_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from address with invalid characters (property-based test).

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

Expect failure.

test_build_send_funds_from_invalid_length_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from address with invalid length (property-based test).

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

Expect failure.

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

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

Expect failure.

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

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

Expect failure.

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

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

Expect failure.

test_build_send_funds_to_invalid_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from payment address to non-existent address (property-based test).

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

Expect failure.

test_build_send_funds_to_invalid_chars_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from payment address to address with invalid characters.

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

Expect failure. Property-based test.

test_build_send_funds_to_invalid_length_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from payment address to address with invalid length.

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

Expect failure. Property-based test.

test_duplicated_tx(cluster: ClusterLib, pool_users: List[PoolUser])[source]

Try to send an identical transaction twice.

Expect failure.

test_invalid_length_utxo_hash(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to use invalid UTxO hash as an input (property-based test).

Expect failure.

test_lower_bound_not_supported(cluster: ClusterLib, pool_users: List[PoolUser])[source]

Try to build a Shelley era TX with an –invalid-before argument.

Expect failure.

test_missing_fee(cluster: ClusterLib, pool_users: List[PoolUser])[source]

Try to build a transaction with a missing –fee parameter.

Expect failure.

test_missing_ttl(cluster: ClusterLib, pool_users: List[PoolUser])[source]

Try to build a Shelley era TX with a missing –ttl (–invalid-hereafter) parameter.

Expect failure.

test_missing_tx_in(cluster: ClusterLib, pool_users: List[PoolUser])[source]

Try to build a transaction with a missing –tx-in parameter.

Expect failure.

test_nonexistent_utxo_hash(cluster: ClusterLib, pool_users: List[PoolUser], use_build_cmd: bool)[source]

Try to use nonexistent UTxO hash as an input.

Expect failure.

test_nonexistent_utxo_ix(cluster: ClusterLib, pool_users: List[PoolUser], use_build_cmd: bool)[source]

Try to use nonexistent UTxO TxIx as an input.

Expect failure.

test_past_ttl(cluster: ClusterLib, pool_users: List[PoolUser], use_build_cmd: bool)[source]

Try to send a transaction with ttl in the past.

Expect failure.

test_pbt_before_negative_overflow(cluster: ClusterLib, pool_users: List[PoolUser], use_build_cmd: bool) → None[source]

Try to send a transaction with negative invalid_before and check for int overflow.

Expect failure.

test_pbt_before_positive_overflow(cluster: ClusterLib, pool_users: List[PoolUser], use_build_cmd: bool) → None[source]

Try to send a transaction with invalid_before > MAX_UINT64.

Check for int overflow. Expect failure.

test_pbt_before_too_high(cluster: ClusterLib, pool_users: List[PoolUser], use_build_cmd: bool) → None[source]

Try to send a transaction with invalid_before > MAX_INT64.

Expect failure.

test_send_funds_from_invalid_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from invalid address (property-based test).

Expect failure.

test_send_funds_from_invalid_chars_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from address with invalid characters (property-based test).

Expect failure.

test_send_funds_from_invalid_length_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from address with invalid length (property-based test).

Expect failure.

test_send_funds_to_invalid_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from payment address to non-existent address (property-based test).

Expect failure.

test_send_funds_to_invalid_chars_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from payment address to address with invalid characters.

Expect failure. Property-based test.

test_send_funds_to_invalid_length_address(cluster: ClusterLib, pool_users: List[PoolUser]) → None[source]

Try to send funds from payment address to address with invalid length.

Expect failure. Property-based test.

test_send_funds_to_reward_address(cluster: ClusterLib, pool_users: List[PoolUser], use_build_cmd: bool)[source]

Try to send funds from payment address to stake address.

Expect failure.

test_send_funds_to_utxo_address(cluster: ClusterLib, pool_users: List[PoolUser], use_build_cmd: bool)[source]

Try to send funds from payment address to UTxO address.

Expect failure.

test_wrong_network_magic(cluster: ClusterLib, pool_users: List[PoolUser])[source]

Try to submit a TX with wrong network magic.

Expect failure.

test_wrong_signing_key(cluster: ClusterLib, pool_users: List[PoolUser])[source]

Try to send a transaction signed with wrong signing key.

Expect failure.

test_wrong_tx_era(cluster: ClusterLib, cluster_wrong_tx_era: ClusterLib, pool_users: List[PoolUser])[source]

Try to send a transaction using TX era > network (cluster) era.

Expect failure.

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 a transaction with amount bellow the minimum lovelace required.

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

Expect failure.

test_build_transfer_negative_amount(cluster: ClusterLib, payment_addrs: List[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Try to build a transaction with negative Lovelace amount.

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

Expect failure.

test_build_transfer_unavailable_funds(cluster: ClusterLib, payment_addrs: List[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Try to build a transaction with more funds than available transaction build.

Check that it is not possible to build such transaction.

test_negative_change(cluster: ClusterLib, payment_addrs: List[AddressRecord])[source]

Try to build a transaction with a negative change.

Check that it is not possible to build such transaction.

test_out_of_bounds_amount(cluster: ClusterLib, payment_addrs: List[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]: Try to build a transaction with output Lovelace amount that is out of bounds.

test_transfer_amount_bellow_minimum(cluster: ClusterLib, payment_addrs: List[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Try to build a transaction with amount bellow the minimum lovelace required.

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

Expect failure.

test_transfer_negative_amount(cluster: ClusterLib, payment_addrs: List[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Try to build a transaction with negative Lovelace amount.

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

Expect failure.

test_wrong_balance(cluster: ClusterLib, payment_addrs: List[AddressRecord], pbt_highest_utxo: UTXOData) → None[source]

Build a transaction with unbalanced change (property-based test).

build an unbalanced transaction
check that it is not possible to submit such transaction

cardano_node_tests.tests.test_update_proposals module

Tests for update proposals.

class cardano_node_tests.tests.test_update_proposals.TestNegativeCostModels[source]

Bases: object

Negative tests for cost models update.

payment_addr(cluster_manager: ClusterManager, cluster: ClusterLib) → AddressRecord[source]: Create new payment address.

pytestmark = [Mark(name='skipif', args=(True,), kwargs={'reason': "Doesn't run on cluster era >= Conway"})][source]

test_incompatible_cost_models(cluster: ClusterLib, payment_addr: AddressRecord, cost_model: str)[source]: Test incompatible Plutus cost models.

class cardano_node_tests.tests.test_update_proposals.TestUpdateProposals[source]

Bases: object

Tests for update proposals.

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

payment_addr(cluster_manager: ClusterManager, cluster_update_proposal: ClusterLib) → AddressRecord[source]: Create new payment address.

pytestmark = [Mark(name='skipif', args=(True,), kwargs={'reason': "Doesn't run on cluster era >= Conway"}), Mark(name='skipif', args=(False,), kwargs={'reason': 'Must run with same cluster and Tx era'})][source]

test_update_proposal(cluster_update_proposal: ClusterLib, payment_addr: AddressRecord)[source]

Test changing protocol parameters using update proposal.

update parameters specific for Alonzo+ eras, and:
- wait for next epoch
- check that parameters were updated
submit update proposal
in the same epoch, submit another update proposal
wait for next epoch
check that parameters were updated with the values submitted in the second update proposal, i.e. the second update proposal overwritten the first one

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_delegation module

cardano_node_tests.tests.test_env_network_id module

cardano_node_tests.tests.test_governance 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_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_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_update_proposals module

cardano_node_tests.tests.test_xdist_helper module

cardano_node_tests.tests.tx_common module

Module contents