Add db-sync config tests

Author: ArturWieczorek

cardano_node_tests/tests/test_dbsync_config.py

@@ -0,0 +1,153 @@
+"""Tests for basic DB-Sync configuration options."""
+
+import logging
+import typing as tp
+
+import pytest
+
+from cardano_node_tests.cluster_management import cluster_management
+from cardano_node_tests.utils import configuration
+from cardano_node_tests.utils import dbsync_utils
+
+LOGGER = logging.getLogger(__name__)
+
+pytestmark = [
+    pytest.mark.skipif(
+        configuration.CLUSTERS_COUNT != 1 and not configuration.HAS_DBSYNC,
+        reason="db-sync config tests can run on a single cluster only",
+    ),
+    pytest.mark.dbsync_config,
+]
+
+
+def check_dbsync_state(expected_state: dict) -> None:
+    """Check the state of db-sync tables and columns against expected conditions.
+
+    Args:
+        expected_state: Dictionary specifying conditions to verify where:
+            - Key format:
+                * "table" for table-level checks
+                * "table.column" for column-level checks
+            - Value format:
+                * "empty" - verify table is empty
+                * "not_empty" - verify table has rows
+                * "exists" - verify table/column exists
+                * "not_exists" - verify table/column doesn't exist
+                * "column_condition:=0" - custom SQL condition
+                * "column_condition:IS NULL" - NULL check condition
+
+    Returns:
+        bool: True if all conditions match, False otherwise
+    """
+    for key, condition in expected_state.items():
+        if "." in key:  # Column-level check
+            table, column = key.split(".", 1)
+            assert condition.startswith("column_condition:"), (
+                f"Invalid column condition format: {condition}"
+            )
+            column_condition = condition.split(":", 1)[1]
+            dbsync_utils.check_column_condition(table, column, column_condition)
+        else:  # Table-level check
+            match condition:
+                case "empty":
+                    assert dbsync_utils.table_empty(key), (
+                        f"Expected {key} to be empty, but it is not."
+                    )
+                case "not_empty":
+                    assert not dbsync_utils.table_empty(key), (
+                        f"Expected {key} to have data, but it is empty."
+                    )
+                case "exists":
+                    assert dbsync_utils.table_exists(key), (
+                        f"Expected {key} to exist, but it does not."
+                    )
+                case "not_exists":
+                    assert not dbsync_utils.table_exists(key), (
+                        f"Expected {key} to NOT exist, but it does."
+                    )
+                case _:
+                    error_msg = f"Unknown table condition '{condition}' for table '{key}'"
+                    raise ValueError(error_msg)
+
+
+@pytest.fixture
+def db_sync_manager(
+    request: pytest.FixtureRequest, cluster_manager: cluster_management.ClusterManager
+) -> dbsync_utils.DBSyncManager:
+    """Provide db-sync manager on a singleton cluster.
+
+    Creates and returns a DBSyncManager instance with locked cluster resources
+    to ensure exclusive access during testing.
+    """
+    cluster_manager.get(lock_resources=[cluster_management.Resources.CLUSTER])
+    return dbsync_utils.DBSyncManager(request)
+
+
+@pytest.mark.order(-1)
+class TestDBSyncConfig:
+    """Basic tests for DB-Sync Config."""
+
+    def test_basic_tx_out(
+        self,
+        db_sync_manager: dbsync_utils.DBSyncManager,
+    ):
+        """Test tx_out option."""
+        db_config = db_sync_manager.get_config_builder()
+
+        # Test tx_out : enable
+        db_sync_manager.restart_with_config(
+            db_config.with_tx_out(value="enable", force_tx_in=False, use_address_table=False)
+        )
+        check_dbsync_state(
+            {
+                "address": "not_exists",
+                "tx_in": "not_empty",
+                "tx_out": "not_empty",
+                "ma_tx_out": "not_empty",
+            }
+        )
+
+        # Test tx_out : disable
+        db_sync_manager.restart_with_config(
+            db_config.with_tx_out(value="disable", force_tx_in=True, use_address_table=True)
+        )
+        check_dbsync_state(
+            {
+                "address": "not_exists",
+                "tx_in": "empty",
+                "tx_out": "empty",
+                "ma_tx_out": "empty",
+                "tx.fee": "column_condition:=0",
+                "redeemer.script_hash": "column_condition:IS NULL",
+            }
+        )
+
+    @pytest.mark.parametrize(
+        ("tx_cbor_value", "expected_state"), [("enable", "not_empty"), ("disable", "empty")]
+    )
+    def test_cbor(
+        self,
+        db_sync_manager: dbsync_utils.DBSyncManager,
+        tx_cbor_value: tp.Literal["enable", "disable"],
+        expected_state: str,
+    ):
+        """Test tx_cbor option with parametrization."""
+        db_config = db_sync_manager.get_config_builder()
+
+        db_sync_manager.restart_with_config(db_config.with_tx_cbor(tx_cbor_value))
+        check_dbsync_state({"tx_cbor": expected_state})
+
+    @pytest.mark.parametrize(
+        ("multi_asset_enable", "expected_state"), [(True, "not_empty"), (False, "empty")]
+    )
+    def test_multi_asset(
+        self,
+        db_sync_manager: dbsync_utils.DBSyncManager,
+        multi_asset_enable: bool,
+        expected_state: str,
+    ):
+        """Test multi_asset option with parametrization."""
+        db_config = db_sync_manager.get_config_builder()
+
+        db_sync_manager.restart_with_config(db_config.with_multi_asset(enable=multi_asset_enable))
+        check_dbsync_state({"multi_asset": expected_state})

cardano_node_tests/utils/dbsync_queries.py

@@ -620,6 +620,27 @@ def execute(query: str, vars: tp.Sequence = ()) -> tp.Iterator[psycopg2.extensio
             cur.close()
 
 
+@contextlib.contextmanager
+def db_transaction() -> tp.Iterator[None]:
+    """Transaction manager that works with connection pool."""
+    conn = None
+    try:
+        conn = dbsync_conn.conn()
+        # Only modify autocommit if not already in a transaction
+        if conn.get_transaction_status() == psycopg2.extensions.TRANSACTION_STATUS_IDLE:
+            conn.autocommit = False
+        yield
+        conn.commit()
+    except Exception:
+        if conn and not conn.closed:
+            conn.rollback()
+        raise
+    finally:
+        if conn and not conn.closed:
+            conn.autocommit = True  # Reset to default psycopg2 behavior
+            # Don't close the connection - let execute() manage it
+
+
 class SchemaVersion:
     """Query and cache db-sync schema version."""
 
@@ -1546,9 +1567,77 @@ def query_drep_distr(
 
 def delete_reserved_pool_tickers() -> int:
     """Delete all records from reserved_pool_ticker and return the number of affected rows."""
-    query = "DELETE FROM reserved_pool_ticker"
+    query = "DELETE FROM reserved_pool_ticker;"
 
     with execute(query=query) as cur:
         affected_rows = cur.rowcount or 0
         dbsync_conn.conn().commit()
         return affected_rows
+
+
+def query_db_sync_progress() -> float:
+    """Calculate blockchain sync percentage (0-100).
+
+    Returns:
+        float: Sync percentage (0-100) if blocks exist
+        None: If no blocks in database
+    """
+    query = (
+        "SELECT"
+        " 100 * (EXTRACT(EPOCH FROM (MAX(time) AT TIME ZONE 'UTC')) -"
+        " EXTRACT(EPOCH FROM (MIN(time) AT TIME ZONE 'UTC')))"
+        " / (EXTRACT(EPOCH FROM (NOW() AT TIME ZONE 'UTC')) -"
+        " EXTRACT(EPOCH FROM (MIN(time) AT TIME ZONE 'UTC')))"
+        " AS sync_percent "
+        " FROM block;"
+    )
+
+    with execute(query=query) as cur:
+        result = cur.fetchone()
+        return min(100.0, float(result[0])) if result else 0.0
+
+
+def query_rows_count(
+    table: str,
+    column: str | None = None,
+    condition: str | None = None,
+    lock: bool = False,
+    lock_timeout: str = "5s",
+) -> int:
+    """Query row count with optional table locking for atomic snapshots.
+
+    Args:
+        table: Table name
+        column: Column name (optional)
+        condition: SQL condition (required if column is provided)
+        lock: Whether to lock table during query
+        lock_timeout: PostgreSQL timeout string (e.g., '5s', '1min')
+
+    Returns:
+        Count of matching rows (0 if none match or table is empty)
+
+    Raises:
+        ValueError: For invalid inputs
+        RuntimeError: If lock times out
+    """
+    # Validation
+    if column is not None and condition is None:
+        error_msg = "Condition required when column is specified."
+        raise ValueError(error_msg)
+
+    # Query construction
+    lock_clause = (
+        f"SET LOCAL lock_timeout = '{lock_timeout}'; LOCK TABLE {table} IN SHARE MODE;"
+        if lock
+        else ""
+    )
+    where_clause = f" WHERE {column} {condition}" if (column and condition) else ""
+    query = f"{lock_clause} SELECT COUNT(*) FROM {table}{where_clause}"
+
+    # Execution
+    with execute(query=query) as cur:
+        try:
+            return cur.fetchone()[0] or 0
+        except psycopg2.errors.LockNotAvailable as e:
+            error_msg = f"Could not acquire lock on {table} within {lock_timeout}"
+            raise RuntimeError(error_msg) from e