Merge pull request #2575 from saratomaz/check_delegaton_vote

Check delegation vote to a DRep in db-sync

cardano_node_tests/tests/tests_conway/test_drep.py

@@ -931,6 +931,7 @@ class TestDelegDReps:
     @submit_utils.PARAM_SUBMIT_METHOD
     @common.PARAM_USE_BUILD_CMD
     @pytest.mark.parametrize("drep", ("always_abstain", "always_no_confidence", "custom"))
+    @pytest.mark.dbsync
     @pytest.mark.testnets
     @pytest.mark.smoke
     def test_dreps_delegation(
@@ -1099,6 +1100,14 @@ def _deregister():
 
         reqc_deleg.success()
 
+        # Check delegation vote on dbsync
+        reqc.db003.start(url=helpers.get_vcs_link())
+        txid = cluster.g_transaction.get_txid(tx_body_file=tx_output.out_file)
+        dbsync_utils.check_delegation_vote(
+            txhash=txid, stake_address=stake_addr_info.address, drep=drep
+        )
+        reqc.db003.success()
+
     @allure.link(helpers.get_vcs_link())
     @submit_utils.PARAM_SUBMIT_METHOD
     @common.PARAM_USE_BUILD_CMD

cardano_node_tests/utils/dbsync_queries.py

@@ -572,6 +572,13 @@ class OffChainVoteDataDBRow:
     vot_anchor_block_id: int
 
 
+@pydantic.dataclasses.dataclass(frozen=True)
+class DelegationVoteDBRow:
+    id: int
+    stake_address_hash_view: str
+    drep_hash_view: str
+
+
 @contextlib.contextmanager
 def execute(query: str, vars: tp.Sequence = ()) -> tp.Iterator[psycopg2.extensions.cursor]:
     # pylint: disable=redefined-builtin
@@ -1463,3 +1470,20 @@ def query_db_size() -> int:
     with execute(query=query) as cur:
         result = cur.fetchone()
         return int(result[0])
+
+
+def query_delegation_vote(txhash: str) -> tp.Generator[DelegationVoteDBRow, None, None]:
+    """Query delegation_vote table in db-sync."""
+    query = (
+        "SELECT"
+        " delegation_vote.id, stake_address.view, drep_hash.view "
+        "FROM delegation_vote "
+        "INNER JOIN stake_address ON stake_address.id = delegation_vote.addr_id "
+        "INNER JOIN drep_hash ON drep_hash.id = delegation_vote.drep_hash_id "
+        "INNER JOIN tx ON tx.id = delegation_vote.tx_id "
+        "WHERE tx.hash = %s;"
+    )
+
+    with execute(query=query, vars=(rf"\x{txhash}",)) as cur:
+        while (result := cur.fetchone()) is not None:
+            yield DelegationVoteDBRow(*result)

cardano_node_tests/utils/dbsync_utils.py

@@ -1484,3 +1484,26 @@ def check_action_data(  # noqa: C901
 
     if errors:
         raise AssertionError("\n".join(errors))
+
+
+def check_delegation_vote(txhash: str, stake_address: str, drep: str) -> None:
+    """Check delegation vote in dbsync."""
+    if not configuration.HAS_DBSYNC:
+        return
+
+    delegation_vote = list(dbsync_queries.query_delegation_vote(txhash=txhash))
+
+    if not delegation_vote:
+        msg = "No delegation vote found in db-sync"
+        raise AssertionError(msg)
+
+    delegation_vote_data = delegation_vote[0]
+
+    assert delegation_vote_data.stake_address_hash_view == stake_address, (
+        "Incorrect delegation vote stake address in dbsync: "
+        f"{delegation_vote_data.stake_address_hash_view} vs {stake_address}"
+    )
+
+    assert delegation_vote_data.stake_address_hash_view == stake_address, (
+        "Incorrect delegation DRep: " f"{delegation_vote_data.drep_hash_view} vs {drep}"
+    )