Add test_register_deregister

Add test that registers and deregisters Plutus script stake address.

cardano_node_tests/tests/tests_plutus/test_delegation.py

@@ -211,6 +211,171 @@ def register_delegate_stake_addr(
     return tx_raw_output, plutus_costs
 
 
+def register_stake_addr(
+    cluster_obj: clusterlib.ClusterLib,
+    temp_template: str,
+    txins: tp.List[clusterlib.UTXOData],
+    collaterals: tp.List[clusterlib.UTXOData],
+    pool_user: delegation.PoolUserScript,
+    redeemer_file: pl.Path,
+    reference_script_utxos: tp.Optional[tp.List[clusterlib.UTXOData]],
+    use_build_cmd: bool,
+) -> tp.Tuple[clusterlib.TxRawOutput, tp.List[dict]]:
+    """Register a stake address."""
+    # Create stake address registration cert
+    stake_addr_reg_cert_file = cluster_obj.g_stake_address.gen_stake_addr_registration_cert(
+        addr_name=f"{temp_template}_addr0", stake_script_file=pool_user.stake.script_file
+    )
+
+    src_init_balance = cluster_obj.g_query.get_address_balance(pool_user.payment.address)
+
+    reg_cert_script = clusterlib.ComplexCert(
+        certificate_file=stake_addr_reg_cert_file,
+        script_file=pool_user.stake.script_file if not reference_script_utxos else "",
+        reference_txin=reference_script_utxos[0] if reference_script_utxos else None,
+        collaterals=collaterals,
+        execution_units=(218855869, 686154),
+        redeemer_file=redeemer_file,
+    )
+
+    tx_files = clusterlib.TxFiles(signing_key_files=[pool_user.payment.skey_file])
+    plutus_costs_reg = []
+
+    if use_build_cmd:
+        tx_raw_output_reg = cluster_obj.g_transaction.build_tx(
+            src_address=pool_user.payment.address,
+            tx_name=f"{temp_template}_reg",
+            txins=txins,
+            tx_files=tx_files,
+            complex_certs=[reg_cert_script],
+            fee_buffer=2_000_000,
+            witness_override=len(tx_files.signing_key_files),
+        )
+        # Calculate cost of Plutus script
+        plutus_costs_reg = cluster_obj.g_transaction.calculate_plutus_script_cost(
+            src_address=pool_user.payment.address,
+            tx_name=f"{temp_template}_reg",
+            txins=txins,
+            tx_files=tx_files,
+            complex_certs=[reg_cert_script],
+            fee_buffer=2_000_000,
+            witness_override=len(tx_files.signing_key_files),
+        )
+        tx_signed_reg = cluster_obj.g_transaction.sign_tx(
+            tx_body_file=tx_raw_output_reg.out_file,
+            signing_key_files=tx_files.signing_key_files,
+            tx_name=f"{temp_template}_reg",
+        )
+        cluster_obj.g_transaction.submit_tx(tx_file=tx_signed_reg, txins=tx_raw_output_reg.txins)
+    else:
+        tx_raw_output_reg = cluster_obj.g_transaction.send_tx(
+            src_address=pool_user.payment.address,
+            tx_name=f"{temp_template}_reg",
+            txins=txins,
+            tx_files=tx_files,
+            complex_certs=[reg_cert_script],
+            fee=300_000,
+        )
+
+    # Check that the balance for source address was correctly updated
+    deposit = cluster_obj.g_query.get_address_deposit()
+    assert (
+        cluster_obj.g_query.get_address_balance(pool_user.payment.address)
+        == src_init_balance - deposit - tx_raw_output_reg.fee
+    ), f"Incorrect balance for source address `{pool_user.payment.address}`"
+
+    # Check that the stake address was registered
+    stake_addr_info = cluster_obj.g_query.get_stake_addr_info(pool_user.stake.address)
+    assert stake_addr_info, f"Stake address was not registered yet: {stake_addr_info}"
+
+    return tx_raw_output_reg, plutus_costs_reg
+
+
+def delegate_stake_addr(
+    cluster_obj: clusterlib.ClusterLib,
+    temp_template: str,
+    txins: tp.List[clusterlib.UTXOData],
+    collaterals: tp.List[clusterlib.UTXOData],
+    pool_user: delegation.PoolUserScript,
+    pool_id: str,
+    redeemer_file: pl.Path,
+    reference_script_utxos: tp.Optional[tp.List[clusterlib.UTXOData]],
+    use_build_cmd: bool,
+) -> tp.Tuple[clusterlib.TxRawOutput, tp.List[dict]]:
+    """Delegate a stake address to a pool."""
+    # Create stake address delegation cert
+    stake_addr_deleg_cert_file = cluster_obj.g_stake_address.gen_stake_addr_delegation_cert(
+        addr_name=f"{temp_template}_addr0",
+        stake_script_file=pool_user.stake.script_file,
+        stake_pool_id=pool_id,
+    )
+
+    src_init_balance = cluster_obj.g_query.get_address_balance(pool_user.payment.address)
+
+    deleg_cert_script = clusterlib.ComplexCert(
+        certificate_file=stake_addr_deleg_cert_file,
+        script_file=pool_user.stake.script_file if not reference_script_utxos else "",
+        reference_txin=reference_script_utxos[0] if reference_script_utxos else None,
+        collaterals=collaterals,
+        execution_units=(218855869, 686154),
+        redeemer_file=redeemer_file,
+    )
+
+    tx_files = clusterlib.TxFiles(signing_key_files=[pool_user.payment.skey_file])
+    plutus_costs_deleg = []
+
+    if use_build_cmd:
+        tx_raw_output_deleg = cluster_obj.g_transaction.build_tx(
+            src_address=pool_user.payment.address,
+            tx_name=f"{temp_template}_deleg",
+            txins=txins,
+            tx_files=tx_files,
+            complex_certs=[deleg_cert_script],
+            fee_buffer=2_000_000,
+            witness_override=len(tx_files.signing_key_files),
+        )
+        # Calculate cost of Plutus script
+        plutus_costs_deleg = cluster_obj.g_transaction.calculate_plutus_script_cost(
+            src_address=pool_user.payment.address,
+            tx_name=f"{temp_template}_deleg",
+            txins=txins,
+            tx_files=tx_files,
+            complex_certs=[deleg_cert_script],
+            fee_buffer=2_000_000,
+            witness_override=len(tx_files.signing_key_files),
+        )
+        tx_signed_deleg = cluster_obj.g_transaction.sign_tx(
+            tx_body_file=tx_raw_output_deleg.out_file,
+            signing_key_files=tx_files.signing_key_files,
+            tx_name=f"{temp_template}_deleg",
+        )
+        cluster_obj.g_transaction.submit_tx(
+            tx_file=tx_signed_deleg, txins=tx_raw_output_deleg.txins
+        )
+    else:
+        tx_raw_output_deleg = cluster_obj.g_transaction.send_tx(
+            src_address=pool_user.payment.address,
+            tx_name=f"{temp_template}_deleg",
+            txins=txins,
+            tx_files=tx_files,
+            complex_certs=[deleg_cert_script],
+            fee=300_000,
+        )
+
+    # Check that the balance for source address was correctly updated
+    assert (
+        cluster_obj.g_query.get_address_balance(pool_user.payment.address)
+        == src_init_balance - tx_raw_output_deleg.fee
+    ), f"Incorrect balance for source address `{pool_user.payment.address}`"
+
+    # Check that the stake address was delegated
+    stake_addr_info = cluster_obj.g_query.get_stake_addr_info(pool_user.stake.address)
+    assert stake_addr_info.delegation, f"Stake address was not delegated yet: {stake_addr_info}"
+    assert pool_id == stake_addr_info.delegation, "Stake address delegated to wrong pool"
+
+    return tx_raw_output_deleg, plutus_costs_deleg
+
+
 def deregister_stake_addr(
     cluster_obj: clusterlib.ClusterLib,
     temp_template: str,
@@ -321,6 +486,165 @@ def deregister_stake_addr(
     return tx_raw_output, plutus_costs
 
 
+# Don't run these tests on testnets as a stake address corresponding to the Plutus script
+# might be already in use.
+@common.SKIPIF_BUILD_UNUSABLE
+@common.PARAM_PLUTUS_VERSION
+@common.PARAM_USE_BUILD_CMD
+class TestRegisterAddr:
+    """Tests for address registration."""
+
+    @allure.link(helpers.get_vcs_link())
+    @pytest.mark.dbsync
+    def test_register_deregister(
+        self,
+        cluster_lock_42stake: tp.Tuple[clusterlib.ClusterLib, str],
+        pool_user: delegation.PoolUserScript,
+        plutus_version: str,
+        use_build_cmd: bool,
+    ):
+        """Register and deregister Plutus script stake address.
+
+        * submit registration certificate for a stake address
+        * check that the stake address was registered
+        * deregister the stake address
+        * check that the key deposit was returned and rewards withdrawn
+        * check that the stake address is no longer registered
+        * (optional) check records in db-sync
+        """
+        # pylint: disable=too-many-locals
+        # It makes no sense to continue if the following issue is still blocking the test
+        if ISSUE_299.is_blocked():
+            ISSUE_299.finish_test()
+
+        __: tp.Any  # mypy workaround
+        cluster, __ = cluster_lock_42stake
+        temp_template = f"{common.get_test_id(cluster)}_{use_build_cmd}_{plutus_version}"
+
+        collateral_fund_reg = 1_500_000_000
+        collateral_fund_withdraw = 1_500_000_000
+        collateral_fund_dereg = 1_500_000_000
+        reg_fund = 1_500_000_000
+        dereg_fund = 1_500_000_000
+
+        if cluster.g_query.get_stake_addr_info(pool_user.stake.address):
+            pytest.skip(
+                f"The Plutus script stake address '{pool_user.stake.address}' is already "
+                "registered, cannot continue."
+            )
+
+        # Step 1: create Tx inputs for step 2 and step 3
+        txouts_step1 = [
+            # For collateral
+            clusterlib.TxOut(address=pool_user.payment.address, amount=collateral_fund_reg),
+            clusterlib.TxOut(address=pool_user.payment.address, amount=collateral_fund_withdraw),
+            clusterlib.TxOut(address=pool_user.payment.address, amount=collateral_fund_dereg),
+            # For registration
+            clusterlib.TxOut(address=pool_user.payment.address, amount=reg_fund),
+            # For deregistration
+            clusterlib.TxOut(address=pool_user.payment.address, amount=dereg_fund),
+        ]
+
+        if plutus_version == "v2":
+            txouts_step1.append(
+                clusterlib.TxOut(
+                    address=pool_user.payment.address,
+                    amount=10_000_000,
+                    reference_script_file=plutus_common.STAKE_PLUTUS_V2,
+                )
+            )
+
+        tx_files_step1 = clusterlib.TxFiles(
+            signing_key_files=[pool_user.payment.skey_file],
+        )
+        tx_output_step1 = cluster.g_transaction.build_tx(
+            src_address=pool_user.payment.address,
+            tx_name=f"{temp_template}_step1",
+            tx_files=tx_files_step1,
+            txouts=txouts_step1,
+            fee_buffer=2_000_000,
+            # Don't join 'change' and 'collateral' txouts, we need separate UTxOs
+            join_txouts=False,
+        )
+        tx_signed_step1 = cluster.g_transaction.sign_tx(
+            tx_body_file=tx_output_step1.out_file,
+            signing_key_files=tx_files_step1.signing_key_files,
+            tx_name=f"{temp_template}_step1",
+        )
+        cluster.g_transaction.submit_tx(tx_file=tx_signed_step1, txins=tx_output_step1.txins)
+
+        step1_utxos = cluster.g_query.get_utxo(tx_raw_output=tx_output_step1)
+        utxo_ix_offset = clusterlib_utils.get_utxo_ix_offset(
+            utxos=step1_utxos, txouts=tx_output_step1.txouts
+        )
+        collateral_reg = clusterlib.filter_utxos(utxos=step1_utxos, utxo_ix=utxo_ix_offset)
+        collateral_withdraw = clusterlib.filter_utxos(utxos=step1_utxos, utxo_ix=utxo_ix_offset + 1)
+        collateral_dereg = clusterlib.filter_utxos(utxos=step1_utxos, utxo_ix=utxo_ix_offset + 2)
+        reg_utxos = clusterlib.filter_utxos(utxos=step1_utxos, utxo_ix=utxo_ix_offset + 3)
+        dereg_utxos = clusterlib.filter_utxos(utxos=step1_utxos, utxo_ix=utxo_ix_offset + 4)
+
+        reference_script_utxos = (
+            clusterlib.filter_utxos(utxos=step1_utxos, utxo_ix=utxo_ix_offset + 5)
+            if plutus_version == "v2"
+            else None
+        )
+
+        # Step 2: register a stake address
+
+        clusterlib_utils.wait_for_epoch_interval(
+            cluster_obj=cluster, start=5, stop=common.EPOCH_STOP_SEC_BUFFER
+        )
+
+        tx_raw_output_reg, plutus_costs_reg = register_stake_addr(
+            cluster_obj=cluster,
+            temp_template=temp_template,
+            txins=reg_utxos,
+            collaterals=collateral_reg,
+            pool_user=pool_user,
+            redeemer_file=plutus_common.REDEEMER_42,
+            reference_script_utxos=reference_script_utxos,
+            use_build_cmd=use_build_cmd,
+        )
+
+        tx_db_record_reg = dbsync_utils.check_tx(
+            cluster_obj=cluster, tx_raw_output=tx_raw_output_reg
+        )
+
+        # Step 3: withdraw rewards and deregister
+
+        reward_error = ""
+
+        # Make sure we have enough time to finish deregistration in one epoch
+        clusterlib_utils.wait_for_epoch_interval(
+            cluster_obj=cluster, start=5, stop=common.EPOCH_STOP_SEC_BUFFER
+        )
+
+        # Submit deregistration certificate and withdraw rewards
+        tx_raw_output_dereg, __ = deregister_stake_addr(
+            cluster_obj=cluster,
+            temp_template=temp_template,
+            txins=dereg_utxos,
+            collaterals=[*collateral_withdraw, *collateral_dereg],
+            pool_user=pool_user,
+            redeemer_file=plutus_common.REDEEMER_42,
+            reference_script_utxos=reference_script_utxos,
+            use_build_cmd=use_build_cmd,
+        )
+
+        if reward_error:
+            raise AssertionError(reward_error)
+
+        # Check tx_view of step 2 and step 3
+        tx_view.check_tx_view(cluster_obj=cluster, tx_raw_output=tx_raw_output_reg)
+        tx_view.check_tx_view(cluster_obj=cluster, tx_raw_output=tx_raw_output_dereg)
+
+        # Compare cost of Plutus script with data from db-sync
+        if tx_db_record_reg and use_build_cmd:
+            dbsync_utils.check_plutus_costs(
+                redeemer_records=tx_db_record_reg.redeemers, cost_records=plutus_costs_reg
+            )
+
+
 # Don't run these tests on testnets as a stake address corresponding to the Plutus script
 # might be already in use.
 @pytest.mark.order(8)