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

MAX_INT_VAL = 18446744073709551616[source]
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.

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

Ignore selected errors in log right after node upgrade.

class cardano_node_tests.tests.test_node_upgrade.TestUpgrade[source]

Bases: object

Tests for node upgrade testing.

pytestmark = [Mark(name='upgrade', args=(), kwargs={})][source]
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])