From df7554aa5aeefc8e6db73c27b5a27fc620053cdb Mon Sep 17 00:00:00 2001
From: saratomaz <sara.tomaz@iohk.io>
Date: Tue, 10 Sep 2024 11:40:26 +0100
Subject: [PATCH] Check drep_distr in dbsync

---
 .../tests/tests_conway/test_drep.py           | 32 +++++++++++++++++--
 cardano_node_tests/utils/dbsync_queries.py    | 27 ++++++++++++++++
 2 files changed, 56 insertions(+), 3 deletions(-)

diff --git a/cardano_node_tests/tests/tests_conway/test_drep.py b/cardano_node_tests/tests/tests_conway/test_drep.py
index 2ad461530..498e3f027 100644
--- a/cardano_node_tests/tests/tests_conway/test_drep.py
+++ b/cardano_node_tests/tests/tests_conway/test_drep.py
@@ -24,6 +24,8 @@
 from cardano_node_tests.utils import blockers
 from cardano_node_tests.utils import cluster_nodes
 from cardano_node_tests.utils import clusterlib_utils
+from cardano_node_tests.utils import configuration
+from cardano_node_tests.utils import dbsync_queries
 from cardano_node_tests.utils import dbsync_utils
 from cardano_node_tests.utils import governance_utils
 from cardano_node_tests.utils import helpers
@@ -1152,13 +1154,15 @@ def _deregister():
         # This takes one epoch, so test this only for selected combinations of build command
         # and submit method, only when we are running on local testnet, and only if we are not
         # running smoke tests.
-        if (
+        check_delegation = (
             use_build_cmd
             and submit_method == submit_utils.SubmitMethods.CLI
             and cluster_nodes.get_cluster_type().type == cluster_nodes.ClusterType.LOCAL
             and "smoke" not in request.config.getoption("-m")
-        ):
-            cluster.wait_for_epoch(epoch_no=init_epoch + 1, padding_seconds=5)
+        )
+
+        if check_delegation:
+            deleg_epoch = cluster.wait_for_epoch(epoch_no=init_epoch + 1, padding_seconds=5)
             deleg_state = clusterlib_utils.get_delegation_state(cluster_obj=cluster)
             stake_addr_hash = cluster.g_stake_address.get_stake_vkey_hash(
                 stake_vkey_file=pool_user_rewards.stake.vkey_file
@@ -1210,6 +1214,28 @@ def _deregister():
         )
         reqc.db003.success()
 
+        # Check DRep distribution in dbsync.
+        # A new entry is created for each DRep for each epoch in drep_distr table.
+        if check_delegation and configuration.HAS_DBSYNC:
+            reqc.db014.start(url=helpers.get_vcs_link())
+            cluster.wait_for_new_epoch()
+            if drep == "custom":
+                drep_id_bech32 = helpers.encode_bech32(prefix="drep", data=drep_id)
+            else:
+                drep_id_bech32 = f"drep_{drep_id}"
+            db_drep_distr = list(
+                dbsync_queries.query_drep_distr(drep_hash=drep_id_bech32, epoch_no=deleg_epoch)
+            )
+
+            assert (
+                db_drep_distr
+            ), f"No Drep distribution found for Drep {drep_id_bech32} and epoch {deleg_epoch}"
+            assert (
+                db_drep_distr[0].amount >= deleg_amount
+            ), f"Unexpected delegated amount in dbsync: {db_drep_distr[0].amount} < {deleg_amount}"
+
+            reqc.db014.success()
+
     @allure.link(helpers.get_vcs_link())
     @submit_utils.PARAM_SUBMIT_METHOD
     @common.PARAM_USE_BUILD_CMD
diff --git a/cardano_node_tests/utils/dbsync_queries.py b/cardano_node_tests/utils/dbsync_queries.py
index 3854e23b2..264db5da1 100644
--- a/cardano_node_tests/utils/dbsync_queries.py
+++ b/cardano_node_tests/utils/dbsync_queries.py
@@ -589,6 +589,15 @@ class NewConstitutionInfoDBRow:
     action_ix: int
 
 
+@pydantic.dataclasses.dataclass(frozen=True)
+class DrepDistributionDBRow:
+    id: int
+    hash_id: int
+    amount: int
+    epoch_no: int
+    drep_hash_view: str
+
+
 @contextlib.contextmanager
 def execute(query: str, vars: tp.Sequence = ()) -> tp.Iterator[psycopg2.extensions.cursor]:
     # pylint: disable=redefined-builtin
@@ -1514,3 +1523,21 @@ def query_new_constitution(txhash: str) -> tp.Generator[NewConstitutionInfoDBRow
     with execute(query=query, vars=(rf"\x{txhash}",)) as cur:
         while (result := cur.fetchone()) is not None:
             yield NewConstitutionInfoDBRow(*result)
+
+
+def query_drep_distr(
+    drep_hash: str, epoch_no: int
+) -> tp.Generator[DrepDistributionDBRow, None, None]:
+    """Query drep voting power in db-sync."""
+    query = (
+        "SELECT"
+        " drep_distr.id, drep_distr.hash_id, drep_distr.amount, drep_distr.epoch_no,"
+        " drep_hash.view "
+        "FROM drep_distr "
+        "INNER JOIN drep_hash ON drep_hash.id = drep_distr.hash_id "
+        "WHERE drep_hash.view = %s AND drep_distr.epoch_no = %s "
+    )
+
+    with execute(query=query, vars=(drep_hash, epoch_no)) as cur:
+        while (result := cur.fetchone()) is not None:
+            yield DrepDistributionDBRow(*result)