Add SMASH service support for testing

ArturWieczorek · ArturWieczorek · commit e175f3eae7a0 · 2025-02-25T13:03:08.000+01:00
diff --git a/cardano_node_tests/cluster_scripts/conway_fast/run-cardano-smash b/cardano_node_tests/cluster_scripts/conway_fast/run-cardano-smash
@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+
+set -uo pipefail
+
+SOCKET_PATH="$(readlink -m "$CARDANO_NODE_SOCKET_PATH")"
+STATE_CLUSTER="${SOCKET_PATH%/*}"
+STATE_CLUSTER_NAME="${STATE_CLUSTER##*/}"
+INSTANCE_NUM="%%INSTANCE_NUM%%"
+
+export PGPASSFILE="$STATE_CLUSTER/pgpass"
+export SMASH_ADMIN="${SMASH_ADMIN:-admin}"
+export SMASH_PASSWORD="${SMASH_PASSWORD:-password}"
+export SMASH_ADMINS_FILE="$STATE_CLUSTER/admins.txt"
+export SMASH_BASE_PORT="${SMASH_BASE_PORT:-31000}"
+export SMASH_PORT="$((SMASH_BASE_PORT + ${INSTANCE_NUM:-0}))"
+
+echo "${SMASH_ADMIN}, ${SMASH_PASSWORD}" > "$SMASH_ADMINS_FILE"
+
+exec "$DBSYNC_REPO/smash-server/bin/cardano-smash-server" --config "./$STATE_CLUSTER_NAME/dbsync-config.yaml" --port "$SMASH_PORT" --admins "$SMASH_ADMINS_FILE"
diff --git a/cardano_node_tests/cluster_scripts/conway_fast/start-cluster b/cardano_node_tests/cluster_scripts/conway_fast/start-cluster
@@ -2,6 +2,7 @@
 
 # controlling environment variables:
 # DBSYNC_REPO - will start and configure db-sync if the value is path to db-sync repository
+# SMASH - if set, will start and configure smash
 # ENABLE_LEGACY - if set, local cluster will use legacy networking
 # MIXED_P2P - if set, local cluster will use P2P for some nodes and legacy topology for others
 # UTXO_BACKEND - 'mem' or 'disk', default is 'mem' (or legacy) if unset
@@ -206,6 +207,23 @@ startsecs=5
 EoF
 fi
 
+# enable smash service
+if [ -n "${DBSYNC_REPO:-""}" ] && [ -n "${SMASH:-""}" ]; then
+  [ -e "${DBSYNC_REPO}/smash-server/bin/cardano-smash-server" ] || \
+    { echo "The \`${DBSYNC_REPO}/smash-server/bin/cardano-smash-server\` not found, line $LINENO" >&2; exit 1; }  # assert
+
+  cat >> "${STATE_CLUSTER}/supervisor.conf" <<EoF
+
+[program:smash]
+command=${SCRIPT_DIR}/run-cardano-smash
+stderr_logfile=./${STATE_CLUSTER_NAME}/smash.stderr
+stdout_logfile=./${STATE_CLUSTER_NAME}/smash.stdout
+autostart=false
+autorestart=false
+startsecs=5
+EoF
+fi
+
 # enable cardano-submit-api service
 if [ "$ENABLE_SUBMIT_API" -eq 1 ]; then
   cat >> "${STATE_CLUSTER}/supervisor.conf" <<EoF
@@ -625,6 +643,12 @@ if [ -n "${DBSYNC_REPO:-""}" ]; then
   supervisorctl -s "unix:///${SUPERVISORD_SOCKET_PATH}" start dbsync
 fi
 
+# start smash
+if [ -n "${DBSYNC_REPO:-""}" ] && [ -n "${SMASH:-""}" ]; then
+  echo "Starting smash"
+  supervisorctl -s "unix:///${SUPERVISORD_SOCKET_PATH}" start smash
+fi
+
 echo "Sleeping for initial Tx submission delay of $TX_SUBMISSION_DELAY seconds"
 sleep "$TX_SUBMISSION_DELAY"
 
diff --git a/cardano_node_tests/tests/test_smash.py b/cardano_node_tests/tests/test_smash.py
@@ -0,0 +1,136 @@
+"""Tests for basic SMASH operations."""
+
+
+import logging
+from http import HTTPStatus
+import pytest
+from cardano_clusterlib import clusterlib
+
+from cardano_node_tests.utils import logfiles
+from cardano_node_tests.utils import configuration
+from cardano_node_tests.utils import dbsync_utils 
+from cardano_node_tests.utils import dbsync_queries
+from cardano_node_tests.utils import smash_utils
+
+LOGGER = logging.getLogger(__name__)
+
+
+@pytest.fixture(autouse=True)
+def check_smash_availability():
+    """Fixture to check SMASH availability before each test."""
+    if not configuration.HAS_SMASH:
+        pytest.skip("Skipping test because SMASH service is not available.")
+
+
+class TestBasicSmash:
+    """Basic tests for SMASH service."""
+
+    @pytest.fixture()
+    def locked_pool(
+        self,
+        cluster_lock_pool: clusterlib.ClusterLib,
+    ) -> dbsync_queries.PoolDataDBRow:
+        """Get pool id from cluster with locked pool."""
+        cluster_obj, pool_name = cluster_lock_pool
+        pools_ids = cluster_obj.g_query.get_stake_pools()
+        locked_pool_number = pool_name[-1]
+        pools = [next(dbsync_queries.query_pool_data(p)) for p in pools_ids]
+        locked_pool_data = next((item for item in pools if 'pool' + locked_pool_number in item.metadata_url), None)
+        return locked_pool_data
+
+    @pytest.fixture(scope="session")
+    def smash(
+        self,
+    ) -> smash_utils.SmashClient:
+        """Create SMASH client."""
+        smash = smash_utils.get_client()
+        return smash
+
+    def test_fetch_pool_metadata(
+            self, 
+            locked_pool: dbsync_queries.PoolDataDBRow,
+            smash: smash_utils.SmashClient
+        ):
+        pool_id = locked_pool.view
+
+        # Offchain metadata is inserted into database few minutes after start of a cluster
+        def _query_func():
+            pool_metadata = next(iter(dbsync_queries.query_off_chain_pool_data(pool_id)), None)
+            assert pool_metadata != None, dbsync_utils.NO_RESPONSE_STR
+            return pool_metadata
+        metadata_dbsync = dbsync_utils.retry_query(query_func=_query_func, timeout=360)
+
+        expected_metadata = smash_utils.PoolMetadata(
+                name=metadata_dbsync.json["name"],
+                description=metadata_dbsync.json["description"],
+                ticker=metadata_dbsync.ticker_name,
+                homepage=metadata_dbsync.json["homepage"]
+            )
+        actual_metadata = smash.get_pool_metadata(pool_id, metadata_dbsync.hash.hex())
+        assert expected_metadata == actual_metadata
+
+    def test_delist_pool(
+            self, 
+            locked_pool: dbsync_queries.PoolDataDBRow,
+            smash: smash_utils.SmashClient,
+            request: pytest.FixtureRequest,
+            worker_id: str,
+        ):
+        pool_id = locked_pool.view
+
+        # Define and register function that ensures pool is re-enlisted after test
+        def pool_cleanup():
+            smash.enlist_pool(pool_id)
+        request.addfinalizer(pool_cleanup)
+
+        # Delist the pool
+        pool_data = dbsync_utils.get_pool_data(pool_id)
+        expected_delisted_pool = smash_utils.PoolData(pool_id=pool_data.hash)
+        actual_delisted_pool = smash.delist_pool(pool_id)
+        assert expected_delisted_pool == actual_delisted_pool
+
+        # Check if fetching metadata for a delisted pool returns an error
+        res_metadata = smash.get_pool_metadata(pool_id, pool_data.metadata_hash)
+        assert isinstance(res_metadata, smash_utils.SmashError)
+        assert HTTPStatus.FORBIDDEN == res_metadata.status_code
+        assert f"Pool {pool_data.hash} is delisted" == res_metadata.message
+
+        # Ignore expected errors in logs that would fail test in teardown phase
+        err_msg = 'Delisted pool already exists!'
+        expected_err_regexes = [err_msg]
+        logfiles.add_ignore_rule(
+            files_glob="smash.stdout",
+            regex="|".join(expected_err_regexes),
+            ignore_file_id=worker_id,
+        )
+        # Ensure re-delisting an already delisted pool returns an error
+        res_delist = smash.delist_pool(pool_id)
+        assert isinstance(res_delist, smash_utils.SmashError)
+        assert HTTPStatus.BAD_REQUEST == res_delist.status_code
+        assert 'DbInsertError' == res_delist.message.get("code")
+        assert err_msg == res_delist.message.get("description")
+
+    def test_enlist_pool(
+            self, 
+            locked_pool: dbsync_queries.PoolDataDBRow,
+            smash: smash_utils.SmashClient,
+        ):
+        pool_id = locked_pool.view
+
+        # Ensure enlisting an already enlisted pool returns an error
+        res_enlist = smash.enlist_pool(pool_id)
+        assert isinstance(res_enlist, smash_utils.SmashError)
+        assert HTTPStatus.NOT_FOUND == res_enlist.status_code
+        assert 'RecordDoesNotExist' == res_enlist.message.get("code")
+        assert 'The requested record does not exist.' == res_enlist.message.get("description")
+
+        # Delist the pool
+        smash.delist_pool(pool_id)
+        pool_data = dbsync_utils.get_pool_data(pool_id)
+        res_metadata = smash.get_pool_metadata(pool_id, pool_data.metadata_hash)
+        assert f"Pool {pool_data.hash} is delisted" == res_metadata.message
+
+        # Enlist pool
+        actual_res_enlist = smash.enlist_pool(pool_id)
+        expected_res_enlist = smash_utils.PoolData(pool_id=pool_data.hash)
+        assert expected_res_enlist == actual_res_enlist
diff --git a/cardano_node_tests/utils/configuration.py b/cardano_node_tests/utils/configuration.py
@@ -35,6 +35,10 @@ def _check_cardano_node_socket_path() -> None:
 # See `cat /proc/sys/net/ipv4/ip_local_port_range`.
 PORTS_BASE = int(os.environ.get("PORTS_BASE") or 23000)
 
+SMASH_BASE_PORT = int(os.environ.get("SMASH_BASE_PORT", "31000"))
+SMASH_ADMIN = os.environ.get("SMASH_ADMIN", "admin")
+SMASH_PASSWORD = os.environ.get("SMASH_PASSWORD", "password")
+
 # Used also in startup scripts as `if [ -n "$VAR" ]...`
 ENABLE_LEGACY = (os.environ.get("ENABLE_LEGACY") or "") != ""
 # Used also in startup scripts as `if [ -n "$VAR" ]...`
@@ -46,6 +50,9 @@ def _check_cardano_node_socket_path() -> None:
 # Number of new blocks before the Tx is considered confirmed. Use default value if set to 0.
 CONFIRM_BLOCKS_NUM = int(os.environ.get("CONFIRM_BLOCKS_NUM") or 0)
 
+# Used also in startup scripts as `if [ -n "$VAR" ]...`
+PV10 = (os.environ.get("PV10") or "") == ""
+
 # Used also in startup scripts
 UTXO_BACKEND = os.environ.get("UTXO_BACKEND") or ""
 if UTXO_BACKEND not in ("", "mem", "disk"):
@@ -100,6 +107,14 @@ def _check_cardano_node_socket_path() -> None:
 else:
     DBSYNC_BIN = pl.Path("/nonexistent")
 
+HAS_SMASH = HAS_DBSYNC and bool(os.environ.get("SMASH"))
+if HAS_SMASH:
+    SMASH_BIN = (
+        pl.Path(os.environ["DBSYNC_REPO"]).expanduser() / "smash-server" / "bin" / "cardano-smash-server"
+    ).resolve()
+else:
+    SMASH_BIN = pl.Path("/nonexistent")
+
 DONT_OVERWRITE_OUTFILES = bool(os.environ.get("DONT_OVERWRITE_OUTFILES"))
 
 # Cluster instances are kept running after tests finish
diff --git a/cardano_node_tests/utils/smash_utils.py b/cardano_node_tests/utils/smash_utils.py
@@ -0,0 +1,159 @@
+import logging
+import typing as tp
+from dataclasses import dataclass
+from functools import lru_cache
+
+import requests
+from http import HTTPStatus
+from requests.auth import HTTPBasicAuth
+
+from cardano_node_tests.utils import cluster_nodes
+from cardano_node_tests.utils import configuration
+
+LOGGER = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, order=True)
+class PoolMetadata:
+    name: str
+    description: str
+    ticker: str
+    homepage: str
+
+
+@dataclass(frozen=True, order=True)
+class PoolData:
+    pool_id: str
+
+
+@dataclass
+class SmashError:
+    """Standardized error response for API failures."""
+    status_code: HTTPStatus
+    message: str
+
+    def to_dict(self) -> tp.Dict[str, tp.Any]:
+        return {"status_code": self.status_code, "message": self.message}
+
+
+class SmashClient:
+    """Utility class for interacting with SMASH via REST API."""
+
+    def __init__(self, instance_num: int) -> None:
+        self.instance_num = instance_num
+        self.port = configuration.SMASH_BASE_PORT + instance_num
+        self.base_url = f"http://localhost:{self.port}"
+        self.auth = self._get_auth()
+
+    def _get_auth(self) -> tp.Optional[HTTPBasicAuth]:
+        """Get Basic Auth credentials if configured."""
+        admin = getattr(configuration, "SMASH_ADMIN", None)
+        password = getattr(configuration, "SMASH_PASSWORD", None)
+        return HTTPBasicAuth(admin, password) if admin and password else None
+
+    def get_pool_metadata(self, pool_id: str, pool_meta_hash: str) -> PoolMetadata | SmashError:
+        """Fetch stake pool metadata from SMASH, returning a `PoolMetadata` dataclass."""
+        url = f"{self.base_url}/api/v1/metadata/{pool_id}/{pool_meta_hash}"
+        try:
+            response = requests.get(url, auth=self.auth)
+            response.raise_for_status()
+            data = response.json()
+            # Ensure required fields exist before creating the dataclass
+            if not all(key in data for key in ("name", "description", "ticker", "homepage")):
+                return None
+            return PoolMetadata(
+                name=data["name"],
+                description=data["description"],
+                ticker=data["ticker"],
+                homepage=data["homepage"]
+            )
+
+        except requests.exceptions.RequestException as e:
+            return SmashError(        
+                status_code=e.response.status_code,
+                message=e.response.text
+            )
+
+    def delist_pool(self, pool_id: str) -> PoolData | SmashError:
+        """Delist a stake pool, returning PoolData on success or a SmashError on failure."""
+        url = f"{self.base_url}/api/v1/delist"
+
+        try:
+            response = requests.patch(url, json={"poolId": pool_id}, auth=self.auth)
+            response.raise_for_status()
+            data = response.json()
+            return PoolData(pool_id=data["poolId"])
+
+        except requests.exceptions.RequestException as err:
+            return SmashError(        
+                status_code=err.response.status_code,
+                message=err.response.json()
+            )
+
+    def enlist_pool(self, pool_id: str) -> bool:
+        """Enlist a stake pool, returning PoolData on success or a SmashError on failure."""
+        url = f"{self.base_url}/api/v1/enlist"
+        try:
+            response = requests.patch(url, json={"poolId": pool_id}, auth=self.auth)
+            response.raise_for_status()
+            data = response.json()
+            return PoolData(pool_id=data["poolId"])
+        except requests.exceptions.RequestException as err:
+            return SmashError(        
+                status_code=err.response.status_code,
+                message=err.response.json()
+            )
+
+    def reserve_ticker(self, ticker_name: str, pool_hash: str) -> bool:
+        """Reserve a ticker for a stake pool."""
+        url = f"{self.base_url}/api/v1/tickers/{ticker_name}"
+        try:
+            response = requests.post(url, json={"poolHash": pool_hash}, auth=self.auth)
+            response.raise_for_status()
+            return True
+        except requests.exceptions.RequestException as err:
+            LOGGER.warning(f"Failed to reserve ticker {ticker_name}: {err}")
+            return False
+
+    def get_pool_errors(self, pool_id: str, from_date: tp.Optional[str] = None) -> tp.Optional[list]:
+        """Fetch errors for a specific stake pool."""
+        url = f"{self.base_url}/api/v1/errors/{pool_id}"
+        params = {"fromDate": from_date} if from_date else None
+        try:
+            response = requests.get(url, params=params, auth=self.auth)
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.RequestException as err:
+            LOGGER.warning(f"Failed to fetch errors for pool {pool_id}: {err}")
+            return None
+
+    def get_retired_pools(self) -> tp.Optional[list]:
+        """Fetch list of retired pools."""
+        url = f"{self.base_url}/api/v1/retired"
+        try:
+            response = requests.get(url, auth=self.auth)
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.RequestException as err:
+            LOGGER.warning(f"Failed to fetch retired pools: {err}")
+            return None
+
+
+class SmashManager:
+
+    instances: tp.ClassVar[dict[int, SmashClient]] = {}
+
+    @classmethod
+    def get_smash_instance(cls) -> SmashClient:
+        """Return a singleton instance of `SmashClient` for the given cluster instance."""
+        instance_num = cluster_nodes.get_instance_num()
+        if instance_num not in cls.instances:
+            cls.instances[instance_num] = SmashClient(instance_num)
+        return cls.instances[instance_num]
+    
+
+def get_client() -> SmashClient | None:
+    """Global access to the SMASH client singleton."""
+    if not configuration.HAS_SMASH:
+        return None
+    return SmashManager.get_smash_instance()
