refactor: update dbsync snapshot test and service usage

- Refactor `get_test_id` to accept either ClusterLib or ClusterManager.
- Update `TestDBSyncSnapshot` to use ClusterManager and instantiate
  DBSyncSnapshotService directly, removing the fixture.
- Move imports to use module import for dbsync_snapshot_service.
- Use `dataclasses` and `typing` consistently in
  dbsync_snapshot_service.py.
- Minor docstring and formatting improvements.

Author: mkoura

cardano_node_tests/tests/common.py

@@ -198,22 +198,30 @@ def unique_time_str() -> str:
     return str(time.time()).replace(".", "")[-8:]
 
 
-def get_test_id(cluster_obj: clusterlib.ClusterLib) -> str:
+def get_test_id(
+    cluster_or_manager: clusterlib.ClusterLib | cluster_management.ClusterManager,
+) -> str:
     """Return unique test ID - function name + assigned cluster instance + random string.
 
     Log the test ID into cluster manager log file.
     """
+    if isinstance(cluster_or_manager, clusterlib.ClusterLib):
+        cid_part = f"_ci{cluster_or_manager.cluster_id}"
+        cm: cluster_management.ClusterManager = cluster_or_manager._cluster_manager  # type: ignore
+    else:
+        cid_part = ""
+        cm = cluster_or_manager
+
+    cinstance = str(cm._cluster_instance_num) if cm._cluster_instance_num != -1 else ""
+
     curr_test = pytest_utils.get_current_test()
     rand_str = clusterlib.get_rand_str(6)
-    test_id = (
-        f"{curr_test.test_function}{curr_test.test_params}_ci{cluster_obj.cluster_id}_{rand_str}"
-    )
+    test_id = f"{curr_test.test_function}{curr_test.test_params}{cid_part}_{rand_str}"
 
     # Log test ID to cluster manager log file - getting test ID happens early
     # after the start of a test, so the log entry can be used for determining
     # time of the test start
-    cm: cluster_management.ClusterManager = cluster_obj._cluster_manager  # type: ignore
-    cm.log(f"c{cm.cluster_instance_num}: got ID `{test_id}` for '{curr_test.full}'")
+    cm.log(f"c{cinstance}: got ID `{test_id}` for '{curr_test.full}'")
 
     return test_id
 

cardano_node_tests/tests/test_dbsync.py

@@ -18,11 +18,10 @@
 from cardano_node_tests.utils import cluster_nodes
 from cardano_node_tests.utils import clusterlib_utils
 from cardano_node_tests.utils import dbsync_queries
+from cardano_node_tests.utils import dbsync_snapshot_service
 from cardano_node_tests.utils import dbsync_utils
 from cardano_node_tests.utils import helpers
 from cardano_node_tests.utils import logfiles
-from cardano_node_tests.utils.dbsync_snapshot_service import DBSyncSnapshotService
-from cardano_node_tests.utils.dbsync_snapshot_service import SnapshotFile
 from cardano_node_tests.utils.versions import VERSIONS
 
 LOGGER = logging.getLogger(__name__)
@@ -290,8 +289,7 @@ def test_reconnect_dbsync(
         cluster_manager: cluster_management.ClusterManager,
         worker_id: str,
     ):
-        """
-        Check that db-sync reconnects to the node after the node is restarted.
+        """Check that db-sync reconnects to the node after the node is restarted.
 
         * restart all nodes of the running cluster
         * submit a transaction
@@ -391,27 +389,28 @@ def test_epoch(self, cluster: clusterlib.ClusterLib):
 class TestDBSyncSnapshot:
     """Tests for db-sync snapshot availability and freshness."""
 
-    @pytest.fixture()
-    def db_sync_snapshots(
-        self,
-    ) -> DBSyncSnapshotService | None:
-        return DBSyncSnapshotService()
-
     @allure.link(helpers.get_vcs_link())
     @pytest.mark.smoke
-    def test_latest_snapshot_freshness(self, db_sync_snapshots: DBSyncSnapshotService):
-        """
-        Check that the latest db-sync snapshot is not older than 5 days.
+    def test_latest_snapshot_freshness(
+        self,
+        cluster_manager: cluster_management.ClusterManager,
+    ):
+        """Check that the latest db-sync snapshot is not older than 5 days.
 
         This test uses the S3 REST API to query the Cardano mainnet snapshot repository
         and verifies that the most recent snapshot is fresh.
         """
+        common.get_test_id(cluster_manager)
+        db_sync_snapshots = dbsync_snapshot_service.DBSyncSnapshotService()
+
         # 1. Find latest version
         latest_version = db_sync_snapshots.get_latest_version()
         LOGGER.info(f"Latest db-sync version: {latest_version}")
 
         # 2. Get latest snapshot for that version
-        latest_snapshot: SnapshotFile = db_sync_snapshots.get_latest_snapshot(latest_version)
+        latest_snapshot: dbsync_snapshot_service.SnapshotFile = (
+            db_sync_snapshots.get_latest_snapshot(latest_version)
+        )
 
         LOGGER.info(f"Latest snapshot: {latest_snapshot.name}")
         LOGGER.info(f"Snapshot date: {latest_snapshot.last_modified.isoformat()}")

cardano_node_tests/utils/dbsync_snapshot_service.py

@@ -1,12 +1,9 @@
+import dataclasses
 import logging
 import re
 import xml.etree.ElementTree as ET
-from dataclasses import dataclass
-from dataclasses import field
 from datetime import datetime
 from datetime import timezone
-from typing import List
-from typing import Tuple
 
 import requests
 
@@ -16,15 +13,15 @@
 S3_NS_URL = "http://s3.amazonaws.com/doc/2006-03-01/"
 
 
-@dataclass
+@dataclasses.dataclass
 class SnapshotFile:
     """Dataclass to hold parsed snapshot file information."""
 
     key: str
     name: str
     last_modified: datetime  # Timezone-aware datetime object
     size: int
-    size_gb: float = field(init=False)
+    size_gb: float = dataclasses.field(init=False)
 
     def __post_init__(self) -> None:
         self.size_gb = self.size / (1024**3)
@@ -40,11 +37,11 @@ def _get_s3_objects(self, prefix: str = "", delimiter: str = "") -> bytes:
         """Fetch XML content from the S3 bucket using REST API."""
         params = {"list-type": "2", "prefix": prefix, "delimiter": delimiter}
 
-        response = requests.get(self.BUCKET_URL, params=params)
+        response = requests.get(self.BUCKET_URL, params=params, timeout=30)
         response.raise_for_status()
         return response.content
 
-    def _parse_s3_xml(self, xml_content: bytes) -> Tuple[List[str], List[SnapshotFile]]:
+    def _parse_s3_xml(self, xml_content: bytes) -> tuple[list[str], list[SnapshotFile]]:
         """Parse S3 XML response using exact namespace search paths with None checks."""
         root = ET.fromstring(xml_content)
         ns_tag = f"{{{S3_NS_URL}}}"
@@ -78,27 +75,10 @@ def _parse_s3_xml(self, xml_content: bytes) -> Tuple[List[str], List[SnapshotFil
                 )
                 continue  # Skip this entry if critical data is missing
 
-            # Use explicit variables to store the text content only if it exists
-            key_text = key_tag.text if key_tag is not None else None
-            modified_text = modified_tag.text if modified_tag is not None else None
-            size_text = size_tag.text if size_tag is not None else None
+            assert key_tag and modified_tag and size_tag  # For type checker
 
-            # Ensure all three critical tags and their text content exist
-            if not all([key_text, modified_text, size_text]):
-                logger.warning(
-                    "Skipping malformed S3 object entry: Missing Key, LastModified, or Size."
-                )
-                continue  # Skip this entry if critical data is missing
-
-            key = key_text
-            last_modified_str = modified_text
-            size_str = size_text
-
-            if last_modified_str is None:
-                continue
-
-            if key is None:
-                continue
+            key = key_tag.text or ""
+            last_modified_str = modified_tag.text or ""
 
             file_date = datetime.strptime(last_modified_str, "%Y-%m-%dT%H:%M:%S.%fZ").replace(
                 tzinfo=timezone.utc
@@ -109,7 +89,7 @@ def _parse_s3_xml(self, xml_content: bytes) -> Tuple[List[str], List[SnapshotFil
                     key=key,
                     name=key.split("/")[-1],
                     last_modified=file_date,
-                    size=int(size_str) if size_str else 0,
+                    size=int(size_tag.text or 0),
                 )
             )