refactor: improve type handling and code structure

Author: ArturWieczorek

cardano_node_tests/tests/test_dbsync_config.py

@@ -2,12 +2,16 @@
 
 import logging
 import typing as tp
+from enum import Enum
 
+import allure
 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_service_manager as db_sync
 from cardano_node_tests.utils import dbsync_utils
+from cardano_node_tests.utils import helpers
 
 LOGGER = logging.getLogger(__name__)
 
@@ -20,7 +24,25 @@
 ]
 
 
-def check_dbsync_state(expected_state: dict) -> None:
+class TableCondition(str, Enum):
+    """Enum for table-level db-sync state conditions."""
+
+    EMPTY = "empty"
+    NOT_EMPTY = "not_empty"
+    EXISTS = "exists"
+    NOT_EXISTS = "not_exists"
+
+
+class ColumnCondition(str, Enum):
+    """Enum for column-level db-sync condition checks."""
+
+    ZERO = "column_condition:=0"
+    IS_NULL = "column_condition:IS NULL"
+
+
+def check_dbsync_state(
+    expected_state: dict[tp.Union[str, db_sync.Table], TableCondition | ColumnCondition],
+) -> None:
     """Check the state of db-sync tables and columns against expected conditions.
 
     Args:
@@ -29,39 +51,38 @@ def check_dbsync_state(expected_state: dict) -> None:
                 * "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
+                * TableCondition enum values for table-level checks
+                * ColumnCondition enum values for column-level checks
 
     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:"), (
+            assert isinstance(condition, ColumnCondition), (
                 f"Invalid column condition format: {condition}"
             )
-            column_condition = condition.split(":", 1)[1]
+            column_condition = condition.value.split(":", 1)[1]
             dbsync_utils.check_column_condition(table, column, column_condition)
         else:  # Table-level check
+            assert isinstance(condition, TableCondition), (
+                f"Invalid table condition format: {condition}"
+            )
             match condition:
-                case "empty":
+                case TableCondition.EMPTY:
                     assert dbsync_utils.table_empty(key), (
                         f"Expected {key} to be empty, but it is not."
                     )
-                case "not_empty":
+                case TableCondition.NOT_EMPTY:
                     assert not dbsync_utils.table_empty(key), (
                         f"Expected {key} to have data, but it is empty."
                     )
-                case "exists":
+                case TableCondition.EXISTS:
                     assert dbsync_utils.table_exists(key), (
                         f"Expected {key} to exist, but it does not."
                     )
-                case "not_exists":
+                case TableCondition.NOT_EXISTS:
                     assert not dbsync_utils.table_exists(key), (
                         f"Expected {key} to NOT exist, but it does."
                     )
@@ -73,81 +94,96 @@ def check_dbsync_state(expected_state: dict) -> None:
 @pytest.fixture
 def db_sync_manager(
     request: pytest.FixtureRequest, cluster_manager: cluster_management.ClusterManager
-) -> dbsync_utils.DBSyncManager:
+) -> db_sync.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)
+    return db_sync.DBSyncManager(request)
 
 
 @pytest.mark.order(-1)
 class TestDBSyncConfig:
     """Basic tests for DB-Sync Config."""
 
+    @allure.link(helpers.get_vcs_link())
+    @pytest.mark.dbsync_config
     def test_basic_tx_out(
         self,
-        db_sync_manager: dbsync_utils.DBSyncManager,
+        db_sync_manager: db_sync.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)
+            db_config.with_tx_out(
+                value=db_sync.TxOutMode.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",
+                db_sync.Table.ADDRESS: TableCondition.NOT_EXISTS,
+                db_sync.Table.TX_IN: TableCondition.NOT_EMPTY,
+                db_sync.Table.TX_OUT: TableCondition.NOT_EMPTY,
+                db_sync.Table.MA_TX_OUT: TableCondition.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)
+            db_config.with_tx_out(
+                value=db_sync.TxOutMode.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",
+                db_sync.Table.ADDRESS: TableCondition.NOT_EXISTS,
+                db_sync.Table.TX_IN: TableCondition.EMPTY,
+                db_sync.Table.TX_OUT: TableCondition.EMPTY,
+                db_sync.Table.MA_TX_OUT: TableCondition.EMPTY,
+                db_sync.Column.Tx.FEE: ColumnCondition.ZERO,
+                db_sync.Column.Redeemer.SCRIPT_HASH: ColumnCondition.IS_NULL,
             }
         )
 
+    @allure.link(helpers.get_vcs_link())
+    @pytest.mark.dbsync_config
     @pytest.mark.parametrize(
-        ("tx_cbor_value", "expected_state"), [("enable", "not_empty"), ("disable", "empty")]
+        ("tx_cbor_value", "expected_state"),
+        [
+            (db_sync.SettingState.ENABLE, TableCondition.NOT_EMPTY),
+            (db_sync.SettingState.DISABLE, TableCondition.EMPTY),
+        ],
     )
     def test_cbor(
         self,
-        db_sync_manager: dbsync_utils.DBSyncManager,
-        tx_cbor_value: tp.Literal["enable", "disable"],
-        expected_state: str,
+        db_sync_manager: db_sync.DBSyncManager,
+        tx_cbor_value: db_sync.SettingState,
+        expected_state: TableCondition,
     ):
         """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})
+        check_dbsync_state({db_sync.Table.TX_CBOR: expected_state})
 
+    @allure.link(helpers.get_vcs_link())
+    @pytest.mark.dbsync_config
     @pytest.mark.parametrize(
-        ("multi_asset_enable", "expected_state"), [(True, "not_empty"), (False, "empty")]
+        ("multi_asset_enable", "expected_state"),
+        [(True, TableCondition.NOT_EMPTY), (False, TableCondition.EMPTY)],
     )
     def test_multi_asset(
         self,
-        db_sync_manager: dbsync_utils.DBSyncManager,
+        db_sync_manager: db_sync.DBSyncManager,
         multi_asset_enable: bool,
-        expected_state: str,
+        expected_state: TableCondition,
     ):
         """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})
+        check_dbsync_state({db_sync.Table.MULTI_ASSET: expected_state})