fix(drep): correct DRep delegation assertion logic

Refactor DRep delegation assertions in tests to use a new helper
`get_drep_cred_name_from_addr_info`, ensuring proper handling of DRep
IDs and key hashes. This improves reliability of delegation checks and
sanitizes DRep ID extraction from stake address info.

Author: mkoura

cardano_node_tests/tests/tests_conway/test_drep.py

@@ -845,9 +845,11 @@ def _deregister():
 
         stake_addr_info = cluster.g_query.get_stake_addr_info(pool_user_rewards.stake.address)
 
-        assert stake_addr_info.vote_delegation == governance_utils.get_drep_cred_name(
-            drep_id=drep1.drep_id
-        ), "Votes are NOT delegated to the correct DRep 1 placed at last of certificates list."
+        assert governance_utils.get_drep_cred_name_from_addr_info(
+            addr_info=stake_addr_info
+        ) == governance_utils.get_drep_cred_name(drep_id=drep1.drep_id), (
+            "Votes are NOT delegated to the correct DRep 1 placed at last of certificates list."
+        )
         reqc.cip087.success()
 
     @allure.link(helpers.get_vcs_link())
@@ -1163,9 +1165,11 @@ def _deregister():
             f"Stake address is NOT registered: {pool_user_rewards.stake.address}"
         )
         reqc.cli035.start(url=helpers.get_vcs_link())
-        assert stake_addr_info.vote_delegation == governance_utils.get_drep_cred_name(
-            drep_id=drep_id
-        ), "Votes are NOT delegated to the correct DRep"
+        assert governance_utils.get_drep_cred_name_from_addr_info(
+            addr_info=stake_addr_info
+        ) == governance_utils.get_drep_cred_name(drep_id=drep_id), (
+            "Votes are NOT delegated to the correct DRep"
+        )
         reqc.cli035.success()
 
         out_utxos = cluster.g_query.get_utxo(tx_raw_output=tx_output)
@@ -1359,9 +1363,11 @@ def _deregister():
         stake_addr_info = cluster.g_query.get_stake_addr_info(pool_user_wpr.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"
-        assert stake_addr_info.vote_delegation == governance_utils.get_drep_cred_name(
-            drep_id=drep_id
-        ), "Votes are NOT delegated to the correct DRep"
+        assert governance_utils.get_drep_cred_name_from_addr_info(
+            addr_info=stake_addr_info
+        ) == governance_utils.get_drep_cred_name(drep_id=drep_id), (
+            "Votes are NOT delegated to the correct DRep"
+        )
 
         # Check the expected balance
         out_utxos = cluster.g_query.get_utxo(tx_raw_output=tx_output)
@@ -1518,9 +1524,11 @@ def _deregister():
         stake_addr_info_deleg1 = cluster.g_query.get_stake_addr_info(
             pool_user_rewards.stake.address
         )
-        assert stake_addr_info_deleg1.vote_delegation == governance_utils.get_drep_cred_name(
-            drep_id=drep1.drep_id
-        ), "Votes are NOT delegated to the correct DRep 1"
+        assert governance_utils.get_drep_cred_name_from_addr_info(
+            addr_info=stake_addr_info_deleg1
+        ) == governance_utils.get_drep_cred_name(drep_id=drep1.drep_id), (
+            "Votes are NOT delegated to the correct DRep 1"
+        )
 
         # Change delegation to second DRep
 
@@ -1549,9 +1557,11 @@ def _deregister():
         stake_addr_info_deleg2 = cluster.g_query.get_stake_addr_info(
             pool_user_rewards.stake.address
         )
-        assert stake_addr_info_deleg2.vote_delegation == governance_utils.get_drep_cred_name(
-            drep_id=drep2.drep_id
-        ), "Votes are NOT changed to the correct DRep 2"
+        assert governance_utils.get_drep_cred_name_from_addr_info(
+            addr_info=stake_addr_info_deleg2
+        ) == governance_utils.get_drep_cred_name(drep_id=drep2.drep_id), (
+            "Votes are NOT changed to the correct DRep 2"
+        )
         reqc.cip086.success()
 
         # Retire the first DRep
@@ -1588,9 +1598,11 @@ def _deregister():
         else:
             if not stake_addr_info_ret.vote_delegation:
                 issues.ledger_4772.finish_test()
-            assert stake_addr_info_ret.vote_delegation == governance_utils.get_drep_cred_name(
-                drep_id=drep2.drep_id
-            ), "Votes are no longer delegated to DRep 2!"
+            assert governance_utils.get_drep_cred_name_from_addr_info(
+                addr_info=stake_addr_info_ret
+            ) == governance_utils.get_drep_cred_name(drep_id=drep2.drep_id), (
+                "Votes are no longer delegated to DRep 2!"
+            )
         reqc.int002.success()
 
 
@@ -1707,9 +1719,11 @@ def _delegate_addr(
             assert stake_addr_info.address, (
                 f"Stake address is NOT registered: {pool_user.stake.address}"
             )
-            assert stake_addr_info.vote_delegation == governance_utils.get_drep_cred_name(
-                drep_id=drep_reg.drep_id
-            ), "Votes are NOT delegated to the correct DRep"
+            assert governance_utils.get_drep_cred_name_from_addr_info(
+                addr_info=stake_addr_info
+            ) == governance_utils.get_drep_cred_name(drep_id=drep_reg.drep_id), (
+                "Votes are NOT delegated to the correct DRep"
+            )
 
             out_utxos = cluster.g_query.get_utxo(tx_raw_output=tx_output)
             assert (

cardano_node_tests/utils/governance_utils.py

@@ -153,6 +153,8 @@ class AnchorData:
 
 
 def get_drep_cred_name(drep_id: str) -> str:
+    if not drep_id:
+        return ""
     cred_name = f"keyHash-{drep_id}"
     if drep_id == "always_abstain":
         cred_name = "alwaysAbstain"
@@ -162,6 +164,28 @@ def get_drep_cred_name(drep_id: str) -> str:
     return cred_name
 
 
+def get_drep_cred_name_from_addr_info(addr_info: clusterlib.StakeAddrInfo) -> str:
+    drep_id_raw = addr_info.vote_delegation_hex or addr_info.vote_delegation or ""
+    if not drep_id_raw:
+        return ""
+
+    # Decode first if working with Bech32 encoded value.
+    if drep_id_raw.startswith("drep1"):
+        drep_id_sanitized = helpers.decode_bech32(bech32=drep_id_raw)
+    else:
+        # The prefix was already present for the hex value in the older version of cardano-cli.
+        drep_id_sanitized = drep_id_raw.removeprefix("keyHash-")
+
+    if len(drep_id_sanitized) > 56:
+        drep_id_sanitized = drep_id_sanitized[-56:]
+
+    # The DRep ID can be either hex string or special names like "alwaysAbstain".
+    cred_name = (
+        f"keyHash-{drep_id_sanitized}" if len(drep_id_sanitized) == 56 else drep_id_sanitized
+    )
+    return cred_name
+
+
 def get_vote_str(vote: clusterlib.Votes) -> str:
     if vote == vote.YES:
         return "VoteYes"