Shards are attribute of StorageNode class

src/frostfs_testlib/hosting/interfaces.py

@@ -219,12 +219,22 @@ class Host(ABC):
         """
 
     @abstractmethod
-    def delete_pilorama(self, service_name: str) -> None:
+    def delete_file(self, file_path: str) -> None:
         """
-        Deletes all pilorama.db files in the node.
+        Deletes file with provided file path
 
         Args:
-            service_name: Name of storage node service.
+            file_path: full path to the file to delete
+
+        """
+
+    @abstractmethod
+    def is_file_exist(self, file_path: str) -> bool:
+        """
+        Checks if file exist
+
+        Args:
+            file_path: full path to the file to check
 
         """
 

src/frostfs_testlib/storage/dataclasses/frostfs_services.py

@@ -3,7 +3,7 @@ import yaml
 from frostfs_testlib.blockchain import RPCClient
 from frostfs_testlib.storage.constants import ConfigAttributes
 from frostfs_testlib.storage.dataclasses.node_base import NodeBase
-
+from frostfs_testlib.storage.dataclasses.shard import Shard
 
 class InnerRing(NodeBase):
     """
@@ -148,6 +148,20 @@ class StorageNode(NodeBase):
     def get_shards_config(self) -> tuple[str, dict]:
         return self.get_config(self.get_shard_config_path())
 
+    def get_shards(self) -> list[Shard]:
+        config = self.get_shards_config()[1]
+        config["storage"]["shard"].pop("default")
+        return [Shard.from_object(shard) for shard in config["storage"]["shard"].values()]
+
+    def get_shards_from_env(self) -> list[Shard]:
+        config = self.get_shards_config()[1]
+        configObj = ConfigObj(StringIO(config))
+
+        pattern = f"{SHARD_PREFIX}\d*"
+        num_shards = len(set(re.findall(pattern, self.get_shards_config())))
+
+        return [Shard.from_config_object(configObj, shard_id) for shard_id in range(num_shards)]
+
     def get_control_endpoint(self) -> str:
         return self._get_attribute(ConfigAttributes.CONTROL_ENDPOINT)
 
@@ -157,6 +171,9 @@ class StorageNode(NodeBase):
     def get_data_directory(self) -> str:
         return self.host.get_data_directory(self.name)
 
+    def get_storage_config(self) -> str:
+        return self.host.get_storage_config(self.name)
+
     def get_http_hostname(self) -> str:
         return self._get_attribute(ConfigAttributes.HTTP_HOSTNAME)
 
@@ -169,8 +186,11 @@ class StorageNode(NodeBase):
     def delete_fstree(self):
         self.host.delete_fstree(self.name)
 
-    def delete_pilorama(self):
-        self.host.delete_pilorama(self.name)
+    def delete_file(self, file_path: str) -> None:
+        self.host.delete_file(file_path)
+
+    def is_file_exist(self, file_path) -> bool:
+        return self.host.is_file_exist(file_path)
 
     def delete_metabase(self):
         self.host.delete_metabase(self.name)

src/frostfs_testlib/storage/dataclasses/shard.py

@@ -0,0 +1,99 @@
+import json
+import pathlib
+import re
+from dataclasses import dataclass
+from io import StringIO
+
+import allure
+import pytest
+import yaml
+from configobj import ConfigObj
+from frostfs_testlib.cli import FrostfsCli
+from frostfs_testlib.resources.cli import CLI_DEFAULT_TIMEOUT
+from frostfs_testlib.resources.common import DEFAULT_WALLET_CONFIG
+
+SHARD_PREFIX = "FROSTFS_STORAGE_SHARD_"
+BLOBSTOR_PREFIX = "_BLOBSTOR_"
+
+
+@dataclass
+class Blobstor:
+    path: str
+    path_type: str
+
+    def __eq__(self, other) -> bool:
+        if not isinstance(other, self.__class__):
+            raise RuntimeError(f"Only two {self.__class__.__name__} instances can be compared")
+        return self.path == other.path and self.path_type == other.path_type
+
+    def __hash__(self):
+        return hash((self.path, self.path_type))
+
+    @staticmethod
+    def from_config_object(section: ConfigObj, shard_id: str, blobstor_id: str):
+        var_prefix = f"{SHARD_PREFIX}{shard_id}{BLOBSTOR_PREFIX}{blobstor_id}"
+        return Blobstor(section.get(f"{var_prefix}_PATH"), section.get(f"{var_prefix}_TYPE"))
+
+
+@dataclass
+class Shard:
+    blobstor: list[Blobstor]
+    metabase: str
+    writecache: str
+    pilorama: str
+
+    def __eq__(self, other) -> bool:
+        if not isinstance(other, self.__class__):
+            raise RuntimeError(f"Only two {self.__class__.__name__} instances can be compared")
+        return (
+            set(self.blobstor) == set(other.blobstor)
+            and self.metabase == other.metabase
+            and self.writecache == other.writecache
+            and self.pilorama == other.pilorama
+        )
+
+    def __hash__(self):
+        return hash((self.metabase, self.writecache))
+
+    @staticmethod
+    def _get_blobstor_count_from_section(config_object: ConfigObj, shard_id: int):
+        pattern = f"{SHARD_PREFIX}{shard_id}{BLOBSTOR_PREFIX}"
+        blobstors = {key[: len(pattern) + 2] for key in config_object.keys() if pattern in key}
+        return len(blobstors)
+
+    @staticmethod
+    def from_config_object(config_object: ConfigObj, shard_id: int):
+        var_prefix = f"{SHARD_PREFIX}{shard_id}"
+
+        blobstor_count = Shard._get_blobstor_count_from_section(config_object, shard_id)
+        blobstors = [
+            Blobstor.from_config_object(config_object, shard_id, blobstor_id) for blobstor_id in range(blobstor_count)
+        ]
+
+        write_cache_enabled = config_object.as_bool(f"{var_prefix}_WRITECACHE_ENABLED")
+
+        return Shard(
+            blobstors,
+            config_object.get(f"{var_prefix}_METABASE_PATH"),
+            config_object.get(f"{var_prefix}_WRITECACHE_PATH") if write_cache_enabled else "",
+        )
+
+    @staticmethod
+    def from_object(shard):
+        metabase = shard["metabase"]["path"] if "path" in shard["metabase"] else shard["metabase"]
+        writecache = shard["writecache"]["path"] if "path" in shard["writecache"] else shard["writecache"]
+
+        # Currently due to issue we need to check if pilorama exists in keys
+        # TODO: make pilorama mandatory after fix
+        if shard.get("pilorama"):
+            pilorama = shard["pilorama"]["path"] if "path" in shard["pilorama"] else shard["pilorama"]
+        else:
+            pilorama = None
+
+        return Shard(
+            blobstor=[Blobstor(path=blobstor["path"], path_type=blobstor["type"]) for blobstor in shard["blobstor"]],
+            metabase=metabase,
+            writecache=writecache,
+            pilorama=pilorama
+        )
+