feat(dbsync): Introduce DBSyncSnapshotService and snapshot freshness test

Introduces new components to cleanly implement the Cardano DB-Sync snapshot freshness check using the S3 REST API.

* **DBSyncSnapshotService:** A new service class responsible for interacting with the IOHK S3 repository, encapsulating API calls and robust XML parsing logic (including handling S3 namespaces and date formats).
* **Clarity:** Uses `dataclass` for snapshot metadata and standard Pytest best practices.

Author: ArturWieczorek

cardano_node_tests/tests/test_dbsync.py

@@ -20,6 +20,9 @@
 from cardano_node_tests.utils import logfiles
 from cardano_node_tests.utils.versions import VERSIONS
 
+from datetime import datetime, timedelta, timezone
+from cardano_node_tests.utils.dbsync_snapshot_service import DBSyncSnapshotService, SnapshotFile
+
 LOGGER = logging.getLogger(__name__)
 
 
@@ -381,3 +384,49 @@ def test_epoch(self, cluster: clusterlib.ClusterLib):
         assert blocks_data_tx_count == epoch_data_tx_count, (
             f"Transactions count don't match between tables for epoch {epoch}"
         )
+
+class TestDBSyncSnapshot:
+    """Tests for db-sync snapshot availability and freshness."""
+
+    @pytest.fixture()
+    def db_sync_snapshots(
+        self,
+    ) -> DBSyncSnapshotService | None:
+        """Create DBSyncSnapshotService client."""
+        snapshot_service = DBSyncSnapshotService()
+        if snapshot_service is None:
+            pytest.skip("DBSyncSnapshotService is not available.")
+        return snapshot_service
+
+    @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.
+        
+        This test uses the S3 REST API to query the Cardano mainnet snapshot repository
+        and verifies that the most recent snapshot is fresh.
+        """
+        
+        # 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)
+        
+        LOGGER.info(f"Latest snapshot: {latest_snapshot.name}")
+        LOGGER.info(f"Snapshot date: {latest_snapshot.last_modified.isoformat()}")
+        LOGGER.info(f"Snapshot size: {latest_snapshot.size_gb:.2f} GB")
+
+        # 3. Perform freshness check
+        five_days_ago = datetime.now(timezone.utc) - timedelta(days=5)
+        
+        assert latest_snapshot.last_modified >= five_days_ago, (
+            f"The latest snapshot is too old. "
+            f"Age: {(datetime.now(timezone.utc) - latest_snapshot.last_modified).days} days. "
+            f"Snapshot date: {latest_snapshot.last_modified.strftime('%Y-%m-%d %H:%M:%S UTC')}, "
+            f"Limit: 5 days ago ({five_days_ago.strftime('%Y-%m-%d %H:%M:%S UTC')})."
+        )
+        
+        LOGGER.info("Success: The latest snapshot is recent (within 5-day limit).")

cardano_node_tests/utils/dbsync_snapshot_service.py

@@ -0,0 +1,111 @@
+import xml.etree.ElementTree as ET
+from datetime import datetime, timezone
+from typing import List, Tuple
+import requests
+import re
+from dataclasses import dataclass, field
+import logging
+
+logger = logging.getLogger(__name__)
+
+# Define the full S3 namespace URL string outside the class for readability
+S3_NS_URL = 'http://s3.amazonaws.com/doc/2006-03-01/'
+
+@dataclass
+class SnapshotFile:
+    """Dataclass to hold parsed snapshot file information."""
+    key: str
+    name: str
+    last_modified: datetime # This will be a timezone-aware datetime object
+    size: int
+    size_gb: float = field(init=False)
+
+    def __post_init__(self):
+        self.size_gb = self.size / (1024**3)
+
+class DBSyncSnapshotService:
+    """Service class to interact with the Cardano DB-Sync S3 repository."""
+    
+    BUCKET_URL: str = "https://update-cardano-mainnet.iohk.io"
+    ROOT_PREFIX: str = "cardano-db-sync/"
+
+    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.raise_for_status()
+        return response.content
+
+    def _parse_s3_xml(self, xml_content: bytes) -> Tuple[List[str], List[SnapshotFile]]:
+            """Parse S3 XML response using exact namespace search paths from working script."""
+            
+            root = ET.fromstring(xml_content)
+            
+            # Use the full namespace string for element search paths (Crucial Fix)
+            NS_TAG = f'{{{S3_NS_URL}}}'
+            
+            # 1. Extract folders (CommonPrefixes)
+            folders = []
+            for prefix in root.findall(f'.//{NS_TAG}CommonPrefixes'):
+                folder_path = prefix.find(f'{NS_TAG}Prefix').text
+                if folder_path and folder_path.endswith('/'):
+                    folder_name = folder_path.strip('/').split('/')[-1]
+                    folders.append(folder_name)
+            
+            # 2. Extract files (Contents)
+            files = []
+            for content in root.findall(f'.//{NS_TAG}Contents'):
+                # Use the full namespaced tag for child elements, matching your working script
+                key = content.find(f'{NS_TAG}Key').text
+                last_modified = content.find(f'{NS_TAG}LastModified').text
+                size = content.find(f'{NS_TAG}Size').text
+                
+                # Parse ISO 8601 date, ensuring it's converted to UTC (tz-aware)
+                file_date = datetime.strptime(last_modified, '%Y-%m-%dT%H:%M:%S.%fZ').replace(tzinfo=timezone.utc)
+                
+                files.append(SnapshotFile(
+                    key=key,
+                    name=key.split('/')[-1],
+                    last_modified=file_date,
+                    size=int(size) if size else 0
+                ))
+            
+            return folders, files
+
+    def get_latest_version(self) -> str:
+        """Finds the numerically latest db-sync version folder."""
+        xml_content = self._get_s3_objects(prefix=self.ROOT_PREFIX, delimiter="/")
+        folders, _ = self._parse_s3_xml(xml_content)
+        
+        version_folders = [f for f in folders if re.match(r'^\d+\.\d+$', f)]
+        
+        if not version_folders:
+            raise RuntimeError("No version folders found in S3 response.")
+        
+        latest_version = sorted(version_folders, key=lambda v: [int(part) for part in v.split('.')])[-1]
+        return latest_version
+
+    def get_latest_snapshot(self, version: str) -> SnapshotFile:
+        """Finds the latest snapshot file for a given version."""
+        version_prefix = f"{self.ROOT_PREFIX}{version}/"
+        xml_content = self._get_s3_objects(prefix=version_prefix)
+        _, files = self._parse_s3_xml(xml_content)
+        
+        # Filter: Revert to the original working filter (.tgz AND 'snapshot')
+        snapshot_files = [
+            f for f in files 
+            if f.name.endswith('.tgz') and 'snapshot' in f.name.lower()
+        ]
+        
+        if not snapshot_files:
+            file_names = [f.name for f in files]
+            logger.warning(f"Files found in S3 response for {version_prefix}: {file_names}")
+            raise RuntimeError(f"No snapshot files found for version {version}. Filtered files: {file_names}")
+        
+        latest_snapshot = max(snapshot_files, key=lambda x: x.last_modified)
+        return latest_snapshot