add hostname

Signed-off-by: m.malygina <m.malygina@yadro.com>

.devenv.hosting.yaml

@@ -1,6 +1,5 @@
 hosts:
 - address: localhost
-  hostname: localhost
   attributes:
     sudo_shell: false
   plugin_name: docker

src/frostfs_testlib/__init__.py

@@ -1,3 +1,3 @@
 __version__ = "2.0.1"
 
-from .fixtures import configure_testlib, hosting, temp_directory
+from .fixtures import configure_testlib, hosting

src/frostfs_testlib/cli/frostfs_adm/morph.py

@@ -350,129 +350,3 @@ class FrostfsAdmMorph(CliCommand):
                 if param not in ["self", "node_netmap_keys"]
             },
         )
-    
-    def add_rule(
-        self,
-        endpoint: str,
-        chain_id: str,
-        target_name: str,
-        target_type: str,
-        rule: Optional[list[str]] = None,
-        path: Optional[str] = None,
-        chain_id_hex: Optional[bool] = None,
-        wallet: Optional[str] = None,
-        address: Optional[str] = None,
-        timeout: Optional[str] = None,
-    ) -> CommandResult:
-        """Drop objects from the node's local storage
-
-        Args:
-        address: Address of wallet account
-        chain-id: Assign ID to the parsed chain
-        chain-id-hex: Flag to parse chain ID as hex
-        endpoint: Remote node control address (as 'multiaddr' or '<host>:<port>')
-        path: Path to encoded chain in JSON or binary format
-        rule: Rule statement
-        target-name: Resource name in APE resource name format
-        target-type: Resource type(container/namespace)
-        timeout: Timeout for an operation (default 15s)
-        wallet: Path to the wallet or binary key
-
-        Returns:
-            Command`s result.
-        """
-        return self._execute(
-            "control add-rule",
-            **{param: value for param, value in locals().items() if param not in ["self"]},
-        )
-
-    def get_rule(
-        self,
-        endpoint: str,
-        chain_id: str,
-        target_name: str,
-        target_type: str,
-        chain_id_hex: Optional[bool] = None,
-        wallet: Optional[str] = None,
-        address: Optional[str] = None,
-        timeout: Optional[str] = None,
-    ) -> CommandResult:
-        """Drop objects from the node's local storage
-
-        Args:
-        address string       Address of wallet account
-        chain-id string      Chain id
-        chain-id-hex         Flag to parse chain ID as hex
-        endpoint string      Remote node control address (as 'multiaddr' or '<host>:<port>')
-        target-name string   Resource name in APE resource name format
-        target-type string   Resource type(container/namespace)
-        timeout duration     Timeout for an operation (default 15s)
-        wallet string        Path to the wallet or binary key
-
-        Returns:
-            Command`s result.
-        """
-        return self._execute(
-            "control get-rule",
-            **{param: value for param, value in locals().items() if param not in ["self"]},
-        )
-
-    def list_rules(
-        self,
-        target_type: str,
-        target_name: Optional[str] = None,
-        rpc_endpoint: Optional[str] = None,
-        wallet: Optional[str] = None,
-        address: Optional[str] = None,
-        timeout: Optional[str] = None,
-    ) -> CommandResult:
-        """Drop objects from the node's local storage
-
-        Args:
-        address: Address of wallet account
-        endpoint: Remote node control address (as 'multiaddr' or '<host>:<port>')
-        target-name: Resource name in APE resource name format
-        target-type: Resource type(container/namespace)
-        timeout: Timeout for an operation (default 15s)
-        wallet: Path to the wallet or binary key
-
-        Returns:
-            Command`s result.
-        """
-        return self._execute(
-            "morph ape list-rule-chains",
-            **{param: value for param, value in locals().items() if param not in ["self"]},
-        )
-    
-    def remove_rule(
-        self,
-        endpoint: str,
-        chain_id: str,
-        target_name: str,
-        target_type: str,
-        all: Optional[bool] = None,
-        chain_id_hex: Optional[bool] = None,
-        wallet: Optional[str] = None,
-        address: Optional[str] = None,
-        timeout: Optional[str] = None,
-    ) -> CommandResult:
-        """Drop objects from the node's local storage
-
-        Args:
-        address: Address of wallet account
-        all: Remove all chains
-        chain-id: Assign ID to the parsed chain
-        chain-id-hex: Flag to parse chain ID as hex
-        endpoint: Remote node control address (as 'multiaddr' or '<host>:<port>')
-        target-name: Resource name in APE resource name format
-        target-type: Resource type(container/namespace)
-        timeout: Timeout for an operation (default 15s)
-        wallet: Path to the wallet or binary key
-
-        Returns:
-            Command`s result.
-        """
-        return self._execute(
-            "control remove-rule",
-            **{param: value for param, value in locals().items() if param not in ["self"]},
-        )

src/frostfs_testlib/cli/frostfs_cli/container.py

@@ -16,8 +16,6 @@ class FrostfsCliContainer(CliCommand):
         basic_acl: Optional[str] = None,
         await_mode: bool = False,
         disable_timestamp: bool = False,
-        force: bool = False,
-        trace: bool = False,
         name: Optional[str] = None,
         nonce: Optional[str] = None,
         policy: Optional[str] = None,
@@ -39,8 +37,6 @@ class FrostfsCliContainer(CliCommand):
             basic_acl: Hex encoded basic ACL value or keywords like 'public-read-write',
                 'private', 'eacl-public-read' (default "private").
             disable_timestamp: Disable timestamp container attribute.
-            force: Skip placement validity check.
-            trace: Generate trace ID and print it.
             name: Container name attribute.
             nonce: UUIDv4 nonce value for container.
             policy: QL-encoded or JSON-encoded placement policy or path to file with it.
@@ -73,7 +69,6 @@ class FrostfsCliContainer(CliCommand):
         ttl: Optional[int] = None,
         xhdr: Optional[dict] = None,
         force: bool = False,
-        trace: bool = False,
     ) -> CommandResult:
         """
         Delete an existing container.
@@ -83,7 +78,6 @@ class FrostfsCliContainer(CliCommand):
             address: Address of wallet account.
             await_mode: Block execution until container is removed.
             cid: Container ID.
-            trace: Generate trace ID and print it.
             force: Do not check whether container contains locks and remove immediately.
             rpc_endpoint: Remote node address (as 'multiaddr' or '<host>:<port>').
             session: Path to a JSON-encoded container session token.
@@ -110,7 +104,6 @@ class FrostfsCliContainer(CliCommand):
         await_mode: bool = False,
         to: Optional[str] = None,
         json_mode: bool = False,
-        trace: bool = False,
         ttl: Optional[int] = None,
         xhdr: Optional[dict] = None,
         timeout: Optional[str] = None,
@@ -123,7 +116,6 @@ class FrostfsCliContainer(CliCommand):
             await_mode: Block execution until container is removed.
             cid: Container ID.
             json_mode: Print or dump container in JSON format.
-            trace: Generate trace ID and print it.
             rpc_endpoint: Remote node address (as 'multiaddr' or '<host>:<port>').
             to: Path to dump encoded container.
             ttl: TTL value in request meta header (default 2).
@@ -163,8 +155,6 @@ class FrostfsCliContainer(CliCommand):
             cid: Container ID.
             rpc_endpoint: Remote node address (as 'multiaddr' or '<host>:<port>').
             to: Path to dump encoded container.
-            json_mode: Print or dump container in JSON format.
-            trace: Generate trace ID and print it.
             session: Path to a JSON-encoded container session token.
             ttl: TTL value in request meta header (default 2).
             wallet: WIF (NEP-2) string or path to the wallet or binary key.
@@ -184,7 +174,6 @@ class FrostfsCliContainer(CliCommand):
     def list(
         self,
         rpc_endpoint: str,
-        name: Optional[str] = None,
         wallet: Optional[str] = None,
         address: Optional[str] = None,
         generate_key: Optional[bool] = None,
@@ -199,13 +188,11 @@ class FrostfsCliContainer(CliCommand):
 
         Args:
             address: Address of wallet account.
-            name: List containers by the attribute name.
             owner: Owner of containers (omit to use owner from private key).
             rpc_endpoint: Remote node address (as 'multiaddr' or '<host>:<port>').
             ttl: TTL value in request meta header (default 2).
             wallet: WIF (NEP-2) string or path to the wallet or binary key.
             xhdr: Dict with request X-Headers.
-            trace: Generate trace ID and print it.
             timeout: Timeout for the operation (default 15s).
             generate_key: Generate a new private key.
 
@@ -221,11 +208,9 @@ class FrostfsCliContainer(CliCommand):
         self,
         rpc_endpoint: str,
         cid: str,
-        bearer: Optional[str] = None,
         wallet: Optional[str] = None,
         address: Optional[str] = None,
         generate_key: Optional[bool] = None,
-        trace: bool = False,
         ttl: Optional[int] = None,
         xhdr: Optional[dict] = None,
         timeout: Optional[str] = None,
@@ -236,12 +221,10 @@ class FrostfsCliContainer(CliCommand):
         Args:
             address: Address of wallet account.
             cid: Container ID.
-            bearer: File with signed JSON or binary encoded bearer token.
             rpc_endpoint: Remote node address (as 'multiaddr' or '<host>:<port>').
             ttl: TTL value in request meta header (default 2).
             wallet: WIF (NEP-2) string or path to the wallet or binary key.
             xhdr: Dict with request X-Headers.
-            trace: Generate trace ID and print it.
             timeout: Timeout for the operation (default 15s).
             generate_key: Generate a new private key.
 
@@ -253,7 +236,6 @@ class FrostfsCliContainer(CliCommand):
             **{param: value for param, value in locals().items() if param not in ["self"]},
         )
 
-    # TODO Deprecated method with 0.42
     def set_eacl(
         self,
         rpc_endpoint: str,
@@ -299,7 +281,6 @@ class FrostfsCliContainer(CliCommand):
         address: Optional[str] = None,
         ttl: Optional[int] = None,
         from_file: Optional[str] = None,
-        trace: bool = False,
         short: Optional[bool] = True,
         xhdr: Optional[dict] = None,
         generate_key: Optional[bool] = None,
@@ -317,7 +298,6 @@ class FrostfsCliContainer(CliCommand):
             from_file: string File path with encoded container
             timeout: duration Timeout for the operation (default 15 s)
             short: shorten the output of node information.
-            trace: Generate trace ID and print it.
             xhdr: Dict with request X-Headers.
             generate_key: Generate a new private key.
 

src/frostfs_testlib/cli/frostfs_cli/object.py

@@ -370,11 +370,11 @@ class FrostfsCliObject(CliCommand):
         self,
         rpc_endpoint: str,
         cid: str,
-        oid: Optional[str] = None,
         wallet: Optional[str] = None,
         address: Optional[str] = None,
         bearer: Optional[str] = None,
         generate_key: Optional[bool] = None,
+        oid: Optional[str] = None,
         trace: bool = False,
         root: bool = False,
         verify_presence_all: bool = False,

src/frostfs_testlib/cli/frostfs_cli/shards.py

@@ -40,7 +40,7 @@ class FrostfsCliShards(CliCommand):
         self,
         endpoint: str,
         mode: str,
-        id: Optional[list[str]] = None,
+        id: Optional[list[str]],
         wallet: Optional[str] = None,
         wallet_password: Optional[str] = None,
         address: Optional[str] = None,

src/frostfs_testlib/fixtures.py

@@ -7,7 +7,7 @@ import yaml
 
 from frostfs_testlib import reporter
 from frostfs_testlib.hosting.hosting import Hosting
-from frostfs_testlib.resources.common import ASSETS_DIR, HOSTING_CONFIG_FILE
+from frostfs_testlib.resources.common import HOSTING_CONFIG_FILE
 from frostfs_testlib.storage import get_service_registry
 
 
@@ -24,16 +24,6 @@ def configure_testlib():
         registry.register_service(svc.name, svc.load())
 
 
-@pytest.fixture(scope="session")
-def temp_directory(configure_testlib):
-    with reporter.step("Prepare tmp directory"):
-        full_path = ASSETS_DIR
-        if not os.path.exists(full_path):
-            os.mkdir(full_path)
-
-    return full_path
-
-
 @pytest.fixture(scope="session")
 def hosting(configure_testlib) -> Hosting:
     with open(HOSTING_CONFIG_FILE, "r") as file:

src/frostfs_testlib/hosting/docker_host.py

@@ -185,12 +185,6 @@ class DockerHost(Host):
     def is_file_exist(self, file_path: str) -> None:
         raise NotImplementedError("Not implemented for docker")
 
-    def wipefs_storage_node_data(self, service_name: str) -> None:
-        raise NotImplementedError("Not implemented for docker")
-
-    def finish_wipefs(self, service_name: str) -> None:
-        raise NotImplementedError("Not implemented for docker")
-
     def delete_storage_node_data(self, service_name: str, cache_only: bool = False) -> None:
         volume_path = self.get_data_directory(service_name)
 
@@ -246,7 +240,7 @@ class DockerHost(Host):
         until: Optional[datetime] = None,
         unit: Optional[str] = None,
         exclude_filter: Optional[str] = None,
-        priority: Optional[str] = None,
+        priority: Optional[str] = None
     ) -> str:
         client = self._get_docker_client()
         filtered_logs = ""

src/frostfs_testlib/hosting/interfaces.py

@@ -178,21 +178,6 @@ class Host(ABC):
             cache_only: To delete cache only.
         """
 
-    @abstractmethod
-    def wipefs_storage_node_data(self, service_name: str) -> None:
-        """Erases all data of the storage node with specified name.
-
-        Args:
-            service_name: Name of storage node service.
-        """
-
-    def finish_wipefs(self, service_name: str) -> None:
-        """Erases all data of the storage node with specified name.
-
-        Args:
-            service_name: Name of storage node service.
-        """
-
     @abstractmethod
     def delete_fstree(self, service_name: str) -> None:
         """
@@ -312,7 +297,7 @@ class Host(ABC):
         until: Optional[datetime] = None,
         unit: Optional[str] = None,
         exclude_filter: Optional[str] = None,
-        priority: Optional[str] = None,
+        priority: Optional[str] = None
     ) -> str:
         """Get logs from host filtered by regex.
 

src/frostfs_testlib/resources/common.py

@@ -51,5 +51,3 @@ CREDENTIALS_CREATE_TIMEOUT = "1m"
 HOSTING_CONFIG_FILE = os.getenv(
     "HOSTING_CONFIG_FILE", os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", ".devenv.hosting.yaml"))
 )
-
-MORE_LOG = os.getenv("MORE_LOG", "1")

src/frostfs_testlib/resources/error_patterns.py

@@ -29,4 +29,3 @@ S3_MALFORMED_XML_REQUEST = "The XML you provided was not well-formed or did not
 RULE_ACCESS_DENIED_CONTAINER = "access to container operation {operation} is denied by access policy engine: Access denied"
 RULE_ACCESS_DENIED_OBJECT = "access to object operation denied: ape denied request: method {operation}: Access denied"
 NO_RULE_FOUND_CONTAINER = "access to container operation {operation} is denied by access policy engine: NoRuleFound"
-NO_RULE_FOUND_OBJECT = "access to object operation denied: ape denied request: method {operation}: NoRuleFound"

src/frostfs_testlib/shell/local_shell.py

@@ -1,18 +1,15 @@
 import logging
 import subprocess
 import tempfile
-from contextlib import nullcontext
 from datetime import datetime
 from typing import IO, Optional
 
 import pexpect
 
 from frostfs_testlib import reporter
-from frostfs_testlib.resources.common import MORE_LOG
 from frostfs_testlib.shell.interfaces import CommandInspector, CommandOptions, CommandResult, Shell
 
 logger = logging.getLogger("frostfs.testlib.shell")
-step_context = reporter.step if MORE_LOG == "1" else nullcontext
 
 
 class LocalShell(Shell):
@@ -31,7 +28,7 @@ class LocalShell(Shell):
         for inspector in [*self.command_inspectors, *extra_inspectors]:
             command = inspector.inspect(original_command, command)
 
-        with step_context(f"Executing command: {command}"):
+        with reporter.step(f"Executing command: {command}"):
             if options.interactive_inputs:
                 return self._exec_interactive(command, options)
             return self._exec_non_interactive(command, options)

src/frostfs_testlib/steps/cli/container.py

@@ -327,6 +327,13 @@ def _parse_cid(output: str) -> str:
     return splitted[1]
 
 
+@reporter.step("Search container by name")
+def search_container_by_name(name: str, node: ClusterNode):
+    resolver_cls = load_plugin("frostfs.testlib.bucket_cid_resolver", node.host.config.product)
+    resolver: BucketContainerResolver = resolver_cls()
+    return resolver.resolve(node, name)
+
+
 @reporter.step("Search for nodes with a container")
 def search_nodes_with_container(
     wallet: WalletInfo,

src/frostfs_testlib/steps/cli/object.py

@@ -15,7 +15,7 @@ from frostfs_testlib.storage.cluster import Cluster, ClusterNode
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
 from frostfs_testlib.testing import wait_for_success
 from frostfs_testlib.utils import json_utils
-from frostfs_testlib.utils.cli_utils import parse_netmap_output
+from frostfs_testlib.utils.cli_utils import parse_cmd_table, parse_netmap_output
 from frostfs_testlib.utils.file_utils import TestFile
 
 logger = logging.getLogger("NeoLogger")
@@ -623,20 +623,25 @@ def head_object(
 
     # If response is Complex Object header, it has `splitId` key
     if "splitId" in decoded.keys():
+        logger.info("decoding split header")
         return json_utils.decode_split_header(decoded)
 
     # If response is Last or Linking Object header,
     # it has `header` dictionary and non-null `split` dictionary
     if "split" in decoded["header"].keys():
         if decoded["header"]["split"]:
+            logger.info("decoding linking object")
             return json_utils.decode_linking_object(decoded)
 
     if decoded["header"]["objectType"] == "STORAGE_GROUP":
+        logger.info("decoding storage group")
         return json_utils.decode_storage_group(decoded)
 
     if decoded["header"]["objectType"] == "TOMBSTONE":
+        logger.info("decoding tombstone")
         return json_utils.decode_tombstone(decoded)
 
+    logger.info("decoding simple header")
     return json_utils.decode_simple_header(decoded)
 
 
@@ -690,13 +695,11 @@ def neo_go_query_height(shell: Shell, endpoint: str) -> dict:
     latest_block = first_line.split(":")
     # taking second line from command's output contain wallet key
     second_line = output.split("\n")[1]
-    if second_line != "":
     validated_state = second_line.split(":")
     return {
         latest_block[0].replace(":", ""): int(latest_block[1]),
         validated_state[0].replace(":", ""): int(validated_state[1]),
     }
-    return {latest_block[0].replace(":", ""): int(latest_block[1])}
 
 
 @wait_for_success()

src/frostfs_testlib/steps/node_management.py

@@ -13,7 +13,6 @@ from frostfs_testlib.resources.common import MORPH_BLOCK_TIME
 from frostfs_testlib.shell import Shell
 from frostfs_testlib.steps.epoch import tick_epoch, wait_for_epochs_align
 from frostfs_testlib.storage.cluster import Cluster, StorageNode
-from frostfs_testlib.testing.test_control import wait_for_success
 from frostfs_testlib.utils import datetime_utils
 
 logger = logging.getLogger("NeoLogger")
@@ -112,7 +111,10 @@ def get_netmap_snapshot(node: StorageNode, shell: Shell) -> str:
     storage_wallet_path = node.get_wallet_path()
 
     cli = FrostfsCli(shell, FROSTFS_CLI_EXEC, config_file=storage_wallet_config)
-    return cli.netmap.snapshot(rpc_endpoint=node.get_rpc_endpoint(), wallet=storage_wallet_path).stdout
+    return cli.netmap.snapshot(
+        rpc_endpoint=node.get_rpc_endpoint(),
+        wallet=storage_wallet_path,
+    ).stdout
 
 
 @reporter.step("Get shard list for {node}")
@@ -200,7 +202,12 @@ def delete_node_data(node: StorageNode) -> None:
 
 
 @reporter.step("Exclude node {node_to_exclude} from network map")
-def exclude_node_from_network_map(node_to_exclude: StorageNode, alive_node: StorageNode, shell: Shell, cluster: Cluster) -> None:
+def exclude_node_from_network_map(
+    node_to_exclude: StorageNode,
+    alive_node: StorageNode,
+    shell: Shell,
+    cluster: Cluster,
+) -> None:
     node_netmap_key = node_to_exclude.get_wallet_public_key()
 
     storage_node_set_status(node_to_exclude, status="offline")
@@ -214,7 +221,12 @@ def exclude_node_from_network_map(node_to_exclude: StorageNode, alive_node: Stor
 
 
 @reporter.step("Include node {node_to_include} into network map")
-def include_node_to_network_map(node_to_include: StorageNode, alive_node: StorageNode, shell: Shell, cluster: Cluster) -> None:
+def include_node_to_network_map(
+    node_to_include: StorageNode,
+    alive_node: StorageNode,
+    shell: Shell,
+    cluster: Cluster,
+) -> None:
     storage_node_set_status(node_to_include, status="online")
 
     # Per suggestion of @fyrchik we need to wait for 2 blocks after we set status and after tick epoch.
@@ -224,7 +236,7 @@ def include_node_to_network_map(node_to_include: StorageNode, alive_node: Storag
     tick_epoch(shell, cluster)
     time.sleep(datetime_utils.parse_time(MORPH_BLOCK_TIME) * 2)
 
-    await_node_in_map(node_to_include, shell, alive_node)
+    check_node_in_map(node_to_include, shell, alive_node)
 
 
 @reporter.step("Check node {node} in network map")
@@ -238,11 +250,6 @@ def check_node_in_map(node: StorageNode, shell: Shell, alive_node: Optional[Stor
     assert node_netmap_key in snapshot, f"Expected node with key {node_netmap_key} to be in network map"
 
 
-@wait_for_success(300, 15, title="Await node {node} in network map")
-def await_node_in_map(node: StorageNode, shell: Shell, alive_node: Optional[StorageNode] = None) -> None:
-    check_node_in_map(node, shell, alive_node)
-
-
 @reporter.step("Check node {node} NOT in network map")
 def check_node_not_in_map(node: StorageNode, shell: Shell, alive_node: Optional[StorageNode] = None) -> None:
     alive_node = alive_node or node
@@ -269,7 +276,12 @@ def wait_for_node_to_be_ready(node: StorageNode) -> None:
 
 
 @reporter.step("Remove nodes from network map trough cli-adm morph command")
-def remove_nodes_from_map_morph(shell: Shell, cluster: Cluster, remove_nodes: list[StorageNode], alive_node: Optional[StorageNode] = None):
+def remove_nodes_from_map_morph(
+    shell: Shell,
+    cluster: Cluster,
+    remove_nodes: list[StorageNode],
+    alive_node: Optional[StorageNode] = None,
+):
     """
     Move node to the Offline state in the candidates list and tick an epoch to update the netmap
     using frostfs-adm
@@ -288,5 +300,9 @@ def remove_nodes_from_map_morph(shell: Shell, cluster: Cluster, remove_nodes: li
 
     if FROSTFS_ADM_EXEC and FROSTFS_ADM_CONFIG_PATH:
         # If frostfs-adm is available, then we tick epoch with it (to be consistent with UAT tests)
-        frostfsadm = FrostfsAdm(shell=remote_shell, frostfs_adm_exec_path=FROSTFS_ADM_EXEC, config_file=FROSTFS_ADM_CONFIG_PATH)
+        frostfsadm = FrostfsAdm(
+            shell=remote_shell,
+            frostfs_adm_exec_path=FROSTFS_ADM_EXEC,
+            config_file=FROSTFS_ADM_CONFIG_PATH,
+        )
         frostfsadm.morph.remove_nodes(node_netmap_keys)

src/frostfs_testlib/steps/s3/s3_helper.py

@@ -7,9 +7,8 @@ from dateutil.parser import parse
 
 from frostfs_testlib import reporter
 from frostfs_testlib.s3 import S3ClientWrapper, VersioningStatus
-from frostfs_testlib.s3.interfaces import BucketContainerResolver
 from frostfs_testlib.shell import Shell
-from frostfs_testlib.steps.cli.container import search_nodes_with_container
+from frostfs_testlib.steps.cli.container import search_container_by_name, search_nodes_with_container
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
 
@@ -176,11 +175,10 @@ def search_nodes_with_bucket(
     wallet: WalletInfo,
     shell: Shell,
     endpoint: str,
-    bucket_container_resolver: BucketContainerResolver,
 ) -> list[ClusterNode]:
     cid = None
     for cluster_node in cluster.cluster_nodes:
-        cid = bucket_container_resolver.resolve(cluster_node, bucket_name)
+        cid = search_container_by_name(name=bucket_name, node=cluster_node)
         if cid:
             break
     nodes_list = search_nodes_with_container(wallet=wallet, cid=cid, shell=shell, endpoint=endpoint, cluster=cluster)

src/frostfs_testlib/storage/constants.py

@@ -12,15 +12,7 @@ class ConfigAttributes:
     REMOTE_WALLET_CONFIG = "remote_wallet_config_path"
     ENDPOINT_DATA_0 = "endpoint_data0"
     ENDPOINT_DATA_1 = "endpoint_data1"
-    ENDPOINT_DATA_0_NS = "endpoint_data0_namespace"
     ENDPOINT_INTERNAL = "endpoint_internal0"
     ENDPOINT_PROMETHEUS = "endpoint_prometheus"
     CONTROL_ENDPOINT = "control_endpoint"
     UN_LOCODE = "un_locode"
-
-
-class PlacementRule:
-    DEFAULT_PLACEMENT_RULE = "REP 2 IN X CBF 1 SELECT 4 FROM * AS X"
-    SINGLE_PLACEMENT_RULE = "REP 1 IN X CBF 1 SELECT 4 FROM * AS X"
-    REP_2_FOR_3_NODES_PLACEMENT_RULE = "REP 2 IN X CBF 1 SELECT 3 FROM * AS X"
-    DEFAULT_EC_PLACEMENT_RULE = "EC 3.1"

src/frostfs_testlib/storage/controllers/cluster_state_controller.py

@@ -14,7 +14,6 @@ from frostfs_testlib.resources.cli import FROSTFS_ADM_CONFIG_PATH, FROSTFS_ADM_E
 from frostfs_testlib.resources.common import MORPH_BLOCK_TIME
 from frostfs_testlib.shell import CommandOptions, Shell, SshConnectionProvider
 from frostfs_testlib.steps.network import IpHelper
-from frostfs_testlib.steps.node_management import include_node_to_network_map, remove_nodes_from_map_morph
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode, S3Gate, StorageNode
 from frostfs_testlib.storage.controllers.disk_controller import DiskController
 from frostfs_testlib.storage.dataclasses.node_base import NodeBase, ServiceClass
@@ -40,7 +39,6 @@ class ClusterStateController:
         self.stopped_nodes: list[ClusterNode] = []
         self.detached_disks: dict[str, DiskController] = {}
         self.dropped_traffic: list[ClusterNode] = []
-        self.excluded_from_netmap: list[StorageNode] = []
         self.stopped_services: set[NodeBase] = set()
         self.cluster = cluster
         self.healthcheck = healthcheck
@@ -309,17 +307,24 @@ class ClusterStateController:
         self.suspended_services = {}
 
     @reporter.step("Drop traffic to {node}, nodes - {block_nodes}")
-    def drop_traffic(self, node: ClusterNode, wakeup_timeout: int, name_interface: str, block_nodes: list[ClusterNode] = None) -> None:
+    def drop_traffic(
+        self,
+        node: ClusterNode,
+        wakeup_timeout: int,
+        name_interface: str,
+        block_nodes: list[ClusterNode] = None,
+    ) -> None:
         list_ip = self._parse_interfaces(block_nodes, name_interface)
         IpHelper.drop_input_traffic_to_node(node, list_ip)
         time.sleep(wakeup_timeout)
         self.dropped_traffic.append(node)
 
     @reporter.step("Start traffic to {node}")
-    def restore_traffic(self, node: ClusterNode) -> None:
+    def restore_traffic(
+        self,
+        node: ClusterNode,
+    ) -> None:
         IpHelper.restore_input_traffic_to_node(node=node)
-        index = self.dropped_traffic.index(node)
-        self.dropped_traffic.pop(index)
 
     @reporter.step("Restore blocked nodes")
     def restore_all_traffic(self):
@@ -403,7 +408,9 @@ class ClusterStateController:
     @reporter.step("Set MaintenanceModeAllowed - {status}")
     def set_maintenance_mode_allowed(self, status: str, cluster_node: ClusterNode) -> None:
         frostfs_adm = FrostfsAdm(
-            shell=cluster_node.host.get_shell(), frostfs_adm_exec_path=FROSTFS_ADM_EXEC, config_file=FROSTFS_ADM_CONFIG_PATH
+            shell=cluster_node.host.get_shell(),
+            frostfs_adm_exec_path=FROSTFS_ADM_EXEC,
+            config_file=FROSTFS_ADM_CONFIG_PATH,
         )
         frostfs_adm.morph.set_config(set_key_value=f"MaintenanceModeAllowed={status}")
 
@@ -444,25 +451,6 @@ class ClusterStateController:
         else:
             assert netmap[0].node_status == status, f"Node status should be '{status}', but was '{netmap[0].node_status}'"
 
-    def remove_node_from_netmap(self, removes_nodes: list[StorageNode]) -> None:
-        alive_storage = list(set(self.cluster.storage_nodes) - set(removes_nodes))[0]
-        remove_nodes_from_map_morph(self.shell, self.cluster, removes_nodes, alive_storage)
-        self.excluded_from_netmap.extend(removes_nodes)
-
-    def include_node_to_netmap(self, include_node: StorageNode, alive_node: StorageNode):
-        include_node_to_network_map(include_node, alive_node, self.shell, self.cluster)
-        self.excluded_from_netmap.pop(self.excluded_from_netmap.index(include_node))
-
-    def include_all_excluded_nodes(self):
-        if not self.excluded_from_netmap:
-            return
-        alive_node = list(set(self.cluster.storage_nodes) - set(self.excluded_from_netmap))[0]
-        if not alive_node:
-            return
-
-        for exclude_node in self.excluded_from_netmap.copy():
-            self.include_node_to_netmap(exclude_node, alive_node)
-
     def _get_cli(
         self, local_shell: Shell, local_wallet: WalletInfo, cluster_node: ClusterNode
     ) -> tuple[FrostfsAdm, FrostfsCli, FrostfsCli]:
@@ -479,7 +467,11 @@ class ClusterStateController:
 
         frostfs_adm = FrostfsAdm(shell=shell, frostfs_adm_exec_path=FROSTFS_ADM_EXEC, config_file=FROSTFS_ADM_CONFIG_PATH)
         frostfs_cli = FrostfsCli(local_shell, FROSTFS_CLI_EXEC, local_wallet.config_path)
-        frostfs_cli_remote = FrostfsCli(shell=shell, frostfs_cli_exec_path=FROSTFS_CLI_EXEC, config_file=wallet_config_path)
+        frostfs_cli_remote = FrostfsCli(
+            shell=shell,
+            frostfs_cli_exec_path=FROSTFS_CLI_EXEC,
+            config_file=wallet_config_path,
+        )
         return frostfs_adm, frostfs_cli, frostfs_cli_remote
 
     def _enable_date_synchronizer(self, cluster_node: ClusterNode):
@@ -542,5 +534,8 @@ class ClusterStateController:
         
     @reporter.step("Get contract by domain - {domain_name}")
     def get_domain_contracts(self, cluster_node: ClusterNode, domain_name: str):
-        frostfs_adm = FrostfsAdm(shell=cluster_node.host.get_shell(), frostfs_adm_exec_path=FROSTFS_ADM_EXEC)
+        frostfs_adm = FrostfsAdm(
+            shell=cluster_node.host.get_shell(),
+            frostfs_adm_exec_path=FROSTFS_ADM_EXEC,
+        )
         return frostfs_adm.morph.dump_hashes(cluster_node.morph_chain.get_http_endpoint(), domain_name).stdout

src/frostfs_testlib/storage/dataclasses/ape.py

@@ -47,8 +47,6 @@ class ConditionType(HumanReadableEnum):
 class ConditionKey(HumanReadableEnum):
     ROLE = '"\\$Actor:role"'
     PUBLIC_KEY = '"\\$Actor:publicKey"'
-    OBJECT_TYPE = '"\\$Object:objectType"'
-    OBJECT_ID = '"\\$Object:objectID"'
 
 
 class MatchType(HumanReadableEnum):
@@ -77,14 +75,6 @@ class Condition:
     def by_key(*args, **kwargs) -> "Condition":
         return Condition(ConditionKey.PUBLIC_KEY, *args, **kwargs)
 
-    @staticmethod
-    def by_object_type(*args, **kwargs) -> "Condition":
-        return Condition(ConditionKey.OBJECT_TYPE, *args, **kwargs)
-
-    @staticmethod
-    def by_object_id(*args, **kwargs) -> "Condition":
-        return Condition(ConditionKey.OBJECT_ID, *args, **kwargs)
-
 
 class Rule:
     def __init__(

src/frostfs_testlib/storage/dataclasses/frostfs_services.py

@@ -39,18 +39,12 @@ class S3Gate(NodeBase):
     def get_endpoint(self) -> str:
         return self._get_attribute(ConfigAttributes.ENDPOINT_DATA_0)
 
-    def get_ns_endpoint(self, ns_name: str) -> str:
-        return self._get_attribute(f"{ConfigAttributes.ENDPOINT_DATA_0}_namespace").format(namespace=ns_name)
-
     def get_all_endpoints(self) -> list[str]:
         return [
             self._get_attribute(ConfigAttributes.ENDPOINT_DATA_0),
             self._get_attribute(ConfigAttributes.ENDPOINT_DATA_1),
         ]
 
-    def get_ns_endpoint(self, ns_name: str) -> str:
-        return self._get_attribute(ConfigAttributes.ENDPOINT_DATA_0_NS).format(namespace=ns_name)
-
     def service_healthcheck(self) -> bool:
         health_metric = "frostfs_s3_gw_state_health"
         output = self.host.get_shell().exec(f"curl -s localhost:8086 | grep {health_metric} | sed 1,2d").stdout

src/frostfs_testlib/storage/dataclasses/storage_object_info.py

@@ -77,19 +77,3 @@ class NodeNetInfo:
     maintenance_mode_allowed: str = None
     eigen_trust_alpha: str = None
     eigen_trust_iterations: str = None
-
-
-@dataclass
-class Chunk:
-    def __init__(self, object_id: str, required_nodes: list, confirmed_nodes: list, ec_parent_object_id: str, ec_index: int) -> None:
-        self.object_id = object_id
-        self.required_nodes = required_nodes
-        self.confirmed_nodes = confirmed_nodes
-        self.ec_parent_object_id = ec_parent_object_id
-        self.ec_index = ec_index
-
-    def __str__(self) -> str:
-        return self.object_id
-
-    def __repr__(self) -> str:
-        return self.object_id

src/frostfs_testlib/storage/grpc_operations/client_wrappers.py

@@ -1,14 +0,0 @@
-from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
-from frostfs_testlib.storage.grpc_operations import interfaces
-from frostfs_testlib.storage.grpc_operations.implementations import container, object
-
-
-class CliClientWrapper(interfaces.GrpcClientWrapper):
-    def __init__(self, cli: FrostfsCli) -> None:
-        self.cli = cli
-        self.object: interfaces.ObjectInterface = object.ObjectOperations(self.cli)
-        self.container: interfaces.ContainerInterface = container.ContainerOperations(self.cli)
-
-
-class RpcClientWrapper(interfaces.GrpcClientWrapper):
-    pass  # The next series

src/frostfs_testlib/storage/grpc_operations/implementations/chunks.py

@@ -1,165 +0,0 @@
-import json
-from typing import Optional
-
-from frostfs_testlib import reporter
-from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
-from frostfs_testlib.resources.cli import CLI_DEFAULT_TIMEOUT
-from frostfs_testlib.storage.cluster import Cluster, ClusterNode
-from frostfs_testlib.storage.controllers.shards_watcher import ShardsWatcher
-from frostfs_testlib.storage.dataclasses.storage_object_info import Chunk, NodeNetmapInfo
-from frostfs_testlib.storage.grpc_operations import interfaces
-from frostfs_testlib.testing.test_control import wait_for_success
-from frostfs_testlib.utils.cli_utils import parse_netmap_output
-
-
-class ChunksOperations(interfaces.ChunksInterface):
-    def __init__(self, cli: FrostfsCli) -> None:
-        self.cli = cli
-
-    @reporter.step("Search node without chunks")
-    def search_node_without_chunks(self, chunks: list[Chunk], cluster: Cluster, endpoint: str = None) -> list[ClusterNode]:
-        if not endpoint:
-            endpoint = cluster.default_rpc_endpoint
-        netmap = parse_netmap_output(self.cli.netmap.snapshot(endpoint, timeout=CLI_DEFAULT_TIMEOUT).stdout)
-        chunks_node_key = []
-        for chunk in chunks:
-            chunks_node_key.extend(chunk.confirmed_nodes)
-        for node_info in netmap.copy():
-            if node_info.node_id in chunks_node_key and node_info in netmap:
-                netmap.remove(node_info)
-        result = []
-        for node_info in netmap:
-            for cluster_node in cluster.cluster_nodes:
-                if node_info.node == cluster_node.host_ip:
-                    result.append(cluster_node)
-        return result
-
-    @reporter.step("Search node with chunk {chunk}")
-    def get_chunk_node(self, cluster: Cluster, chunk: Chunk) -> tuple[ClusterNode, NodeNetmapInfo]:
-        netmap = parse_netmap_output(self.cli.netmap.snapshot(cluster.default_rpc_endpoint, timeout=CLI_DEFAULT_TIMEOUT).stdout)
-        for node_info in netmap:
-            if node_info.node_id in chunk.confirmed_nodes:
-                for cluster_node in cluster.cluster_nodes:
-                    if cluster_node.host_ip == node_info.node:
-                        return (cluster_node, node_info)
-
-    @wait_for_success(300, 5, fail_testcase=None)
-    @reporter.step("Search shard with chunk {chunk}")
-    def get_shard_chunk(self, node: ClusterNode, chunk: Chunk) -> str:
-        oid_path = f"{chunk.object_id[0]}/{chunk.object_id[1]}/{chunk.object_id[2]}/{chunk.object_id[3]}"
-        node_shell = node.storage_node.host.get_shell()
-        shards_watcher = ShardsWatcher(node)
-
-        with reporter.step("Search object file"):
-            for shard_id, shard_info in shards_watcher.shards_snapshots[-1].items():
-                check_dir = node_shell.exec(f" [ -d {shard_info['blobstor'][1]['path']}/{oid_path} ] && echo 1 || echo 0").stdout
-                if "1" in check_dir.strip():
-                    return shard_id
-
-    @reporter.step("Get all chunks")
-    def get_all(
-        self,
-        rpc_endpoint: str,
-        cid: str,
-        oid: str,
-        address: Optional[str] = None,
-        bearer: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        trace: bool = True,
-        root: bool = False,
-        verify_presence_all: bool = False,
-        json: bool = True,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = None,
-    ) -> list[Chunk]:
-        object_nodes = self.cli.object.nodes(
-            rpc_endpoint=rpc_endpoint,
-            cid=cid,
-            address=address,
-            bearer=bearer,
-            generate_key=generate_key,
-            oid=oid,
-            trace=trace,
-            root=root,
-            verify_presence_all=verify_presence_all,
-            json=json,
-            ttl=ttl,
-            xhdr=xhdr,
-            timeout=timeout,
-        )
-        return self._parse_object_nodes(object_nodes.stdout.split("\n")[0])
-
-    @reporter.step("Get last parity chunk")
-    def get_parity(
-        self,
-        rpc_endpoint: str,
-        cid: str,
-        address: Optional[str] = None,
-        bearer: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        oid: Optional[str] = None,
-        trace: bool = True,
-        root: bool = False,
-        verify_presence_all: bool = False,
-        json: bool = True,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = None,
-    ) -> Chunk:
-        object_nodes = self.cli.object.nodes(
-            rpc_endpoint=rpc_endpoint,
-            cid=cid,
-            address=address,
-            bearer=bearer,
-            generate_key=generate_key,
-            oid=oid,
-            trace=trace,
-            root=root,
-            verify_presence_all=verify_presence_all,
-            json=json,
-            ttl=ttl,
-            xhdr=xhdr,
-            timeout=timeout,
-        )
-        return self._parse_object_nodes(object_nodes.stdout.split("\n")[0])[-1]
-
-    @reporter.step("Get first data chunk")
-    def get_first_data(
-        self,
-        rpc_endpoint: str,
-        cid: str,
-        oid: Optional[str] = None,
-        address: Optional[str] = None,
-        bearer: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        trace: bool = True,
-        root: bool = False,
-        verify_presence_all: bool = False,
-        json: bool = True,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> Chunk:
-        object_nodes = self.cli.object.nodes(
-            rpc_endpoint=rpc_endpoint,
-            cid=cid,
-            address=address,
-            bearer=bearer,
-            generate_key=generate_key,
-            oid=oid,
-            trace=trace,
-            root=root,
-            verify_presence_all=verify_presence_all,
-            json=json,
-            ttl=ttl,
-            xhdr=xhdr,
-            timeout=timeout,
-        )
-        return self._parse_object_nodes(object_nodes.stdout.split("\n")[0])[0]
-
-    def _parse_object_nodes(self, object_nodes: str) -> list[Chunk]:
-        parse_result = json.loads(object_nodes)
-        if parse_result.get("errors"):
-            raise parse_result["errors"]
-        return [Chunk(**chunk) for chunk in parse_result["data_objects"]]

src/frostfs_testlib/storage/grpc_operations/implementations/container.py

@@ -1,330 +0,0 @@
-import json
-import logging
-import re
-from typing import List, Optional, Union
-
-from frostfs_testlib import reporter
-from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
-from frostfs_testlib.plugins import load_plugin
-from frostfs_testlib.resources.cli import CLI_DEFAULT_TIMEOUT
-from frostfs_testlib.s3.interfaces import BucketContainerResolver
-from frostfs_testlib.storage.cluster import Cluster, ClusterNode
-from frostfs_testlib.storage.grpc_operations import interfaces
-from frostfs_testlib.utils import json_utils
-
-logger = logging.getLogger("NeoLogger")
-
-
-class ContainerOperations(interfaces.ContainerInterface):
-    def __init__(self, cli: FrostfsCli) -> None:
-        self.cli = cli
-
-    @reporter.step("Create Container")
-    def create(
-        self,
-        endpoint: str,
-        nns_zone: Optional[str] = None,
-        nns_name: Optional[str] = None,
-        address: Optional[str] = None,
-        attributes: Optional[dict] = None,
-        basic_acl: Optional[str] = None,
-        await_mode: bool = False,
-        disable_timestamp: bool = False,
-        force: bool = False,
-        trace: bool = False,
-        name: Optional[str] = None,
-        nonce: Optional[str] = None,
-        policy: Optional[str] = None,
-        session: Optional[str] = None,
-        subnet: Optional[str] = None,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> str:
-        """
-        A wrapper for `frostfs-cli container create` call.
-
-        Args:
-            wallet (WalletInfo): a wallet on whose behalf a container is created
-            rule (optional, str): placement rule for container
-            basic_acl (optional, str): an ACL for container, will be
-                                appended to `--basic-acl` key
-            attributes (optional, dict): container attributes , will be
-                                appended to `--attributes` key
-            session_token (optional, str): a path to session token file
-            session_wallet(optional, str): a path to the wallet which signed
-                                the session token; this parameter makes sense
-                                when paired with `session_token`
-            shell: executor for cli command
-            endpoint: FrostFS endpoint to send request to, appends to `--rpc-endpoint` key
-            options (optional, dict): any other options to pass to the call
-            name (optional, str): container name attribute
-            await_mode (bool): block execution until container is persisted
-            wait_for_creation (): Wait for container shows in container list
-            timeout: Timeout for the operation.
-
-        Returns:
-            (str): CID of the created container
-        """
-        result = self.cli.container.create(
-            rpc_endpoint=endpoint,
-            policy=policy,
-            nns_zone=nns_zone,
-            nns_name=nns_name,
-            address=address,
-            attributes=attributes,
-            basic_acl=basic_acl,
-            await_mode=await_mode,
-            disable_timestamp=disable_timestamp,
-            force=force,
-            trace=trace,
-            name=name,
-            nonce=nonce,
-            session=session,
-            subnet=subnet,
-            ttl=ttl,
-            xhdr=xhdr,
-            timeout=timeout,
-        )
-
-        cid = self._parse_cid(result.stdout)
-
-        logger.info("Container created; waiting until it is persisted in the sidechain")
-
-        return cid
-
-    @reporter.step("List Containers")
-    def list(
-        self,
-        endpoint: str,
-        name: Optional[str] = None,
-        address: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        owner: Optional[str] = None,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-        **params,
-    ) -> List[str]:
-        """
-        A wrapper for `frostfs-cli container list` call. It returns all the
-        available containers for the given wallet.
-        Args:
-            shell: executor for cli command
-            endpoint: FrostFS endpoint to send request to, appends to `--rpc-endpoint` key
-            timeout: Timeout for the operation.
-        Returns:
-            (list): list of containers
-        """
-        result = self.cli.container.list(
-            rpc_endpoint=endpoint,
-            name=name,
-            address=address,
-            generate_key=generate_key,
-            owner=owner,
-            ttl=ttl,
-            xhdr=xhdr,
-            timeout=timeout,
-            **params,
-        )
-        return result.stdout.split()
-
-    @reporter.step("List Objects in container")
-    def list_objects(
-        self,
-        endpoint: str,
-        cid: str,
-        bearer: Optional[str] = None,
-        wallet: Optional[str] = None,
-        address: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        trace: bool = False,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> List[str]:
-        """
-        A wrapper for `frostfs-cli container list-objects` call. It returns all the
-        available objects in container.
-        Args:
-            container_id: cid of container
-            endpoint: FrostFS endpoint to send request to, appends to `--rpc-endpoint` key
-            timeout: Timeout for the operation.
-        Returns:
-            (list): list of containers
-        """
-        result = self.cli.container.list_objects(
-            rpc_endpoint=endpoint,
-            cid=cid,
-            bearer=bearer,
-            wallet=wallet,
-            address=address,
-            generate_key=generate_key,
-            trace=trace,
-            ttl=ttl,
-            xhdr=xhdr,
-            timeout=timeout,
-        )
-        logger.info(f"Container objects: \n{result}")
-        return result.stdout.split()
-
-    @reporter.step("Delete container")
-    def delete(
-        self,
-        endpoint: str,
-        cid: str,
-        address: Optional[str] = None,
-        await_mode: bool = False,
-        session: Optional[str] = None,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        force: bool = False,
-        trace: bool = False,
-    ):
-        try:
-            return self.cli.container.delete(
-                rpc_endpoint=endpoint,
-                cid=cid,
-                address=address,
-                await_mode=await_mode,
-                session=session,
-                ttl=ttl,
-                xhdr=xhdr,
-                force=force,
-                trace=trace,
-            ).stdout
-        except RuntimeError as e:
-            print(f"Error request:\n{e}")
-
-    @reporter.step("Get container")
-    def get(
-        self,
-        endpoint: str,
-        cid: str,
-        address: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        await_mode: bool = False,
-        to: Optional[str] = None,
-        json_mode: bool = True,
-        trace: bool = False,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> Union[dict, str]:
-        result = self.cli.container.get(
-            rpc_endpoint=endpoint,
-            cid=cid,
-            address=address,
-            generate_key=generate_key,
-            await_mode=await_mode,
-            to=to,
-            json_mode=json_mode,
-            trace=trace,
-            ttl=ttl,
-            xhdr=xhdr,
-            timeout=timeout,
-        )
-        container_info = json.loads(result.stdout)
-        attributes = dict()
-        for attr in container_info["attributes"]:
-            attributes[attr["key"]] = attr["value"]
-        container_info["attributes"] = attributes
-        container_info["ownerID"] = json_utils.json_reencode(container_info["ownerID"]["value"])
-        return container_info
-
-    @reporter.step("Get eacl container")
-    def get_eacl(
-        self,
-        endpoint: str,
-        cid: str,
-        address: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        await_mode: bool = False,
-        json_mode: bool = True,
-        trace: bool = False,
-        to: Optional[str] = None,
-        session: Optional[str] = None,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ):
-        return self.cli.container.get_eacl(
-            rpc_endpoint=endpoint,
-            cid=cid,
-            address=address,
-            generate_key=generate_key,
-            await_mode=await_mode,
-            to=to,
-            session=session,
-            ttl=ttl,
-            xhdr=xhdr,
-            timeout=CLI_DEFAULT_TIMEOUT,
-        ).stdout
-
-    @reporter.step("Get nodes container")
-    def nodes(
-        self,
-        endpoint: str,
-        cid: str,
-        cluster: Cluster,
-        address: Optional[str] = None,
-        ttl: Optional[int] = None,
-        from_file: Optional[str] = None,
-        trace: bool = False,
-        short: Optional[bool] = True,
-        xhdr: Optional[dict] = None,
-        generate_key: Optional[bool] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> List[ClusterNode]:
-        result = self.cli.container.search_node(
-            rpc_endpoint=endpoint,
-            cid=cid,
-            address=address,
-            ttl=ttl,
-            from_file=from_file,
-            trace=trace,
-            short=short,
-            xhdr=xhdr,
-            generate_key=generate_key,
-            timeout=timeout,
-        ).stdout
-
-        pattern = r"[0-9]+(?:\.[0-9]+){3}"
-        nodes_ip = list(set(re.findall(pattern, result)))
-
-        with reporter.step(f"nodes ips = {nodes_ip}"):
-            nodes_list = cluster.get_nodes_by_ip(nodes_ip)
-
-        with reporter.step(f"Return nodes - {nodes_list}"):
-            return nodes_list
-
-    @reporter.step("Resolve container by name")
-    def resolve_container_by_name(name: str, node: ClusterNode):
-        resolver_cls = load_plugin("frostfs.testlib.bucket_cid_resolver", node.host.config.product)
-        resolver: BucketContainerResolver = resolver_cls()
-        return resolver.resolve(node, name)
-
-    def _parse_cid(self, output: str) -> str:
-        """
-        Parses container ID from a given CLI output. The input string we expect:
-                container ID: 2tz86kVTDpJxWHrhw3h6PbKMwkLtBEwoqhHQCKTre1FN
-                awaiting...
-                container has been persisted on sidechain
-        We want to take 'container ID' value from the string.
-
-        Args:
-            output (str): CLI output to parse
-
-        Returns:
-            (str): extracted CID
-        """
-        try:
-            # taking first line from command's output
-            first_line = output.split("\n")[0]
-        except Exception:
-            first_line = ""
-            logger.error(f"Got empty output: {output}")
-        splitted = first_line.split(": ")
-        if len(splitted) != 2:
-            raise ValueError(f"no CID was parsed from command output: \t{first_line}")
-        return splitted[1]

src/frostfs_testlib/storage/grpc_operations/implementations/object.py

@@ -1,624 +0,0 @@
-import json
-import logging
-import os
-import re
-import uuid
-from typing import Any, Optional
-
-from frostfs_testlib import reporter, utils
-from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
-from frostfs_testlib.resources.cli import CLI_DEFAULT_TIMEOUT
-from frostfs_testlib.resources.common import ASSETS_DIR
-from frostfs_testlib.shell.interfaces import CommandResult
-from frostfs_testlib.storage.cluster import Cluster, ClusterNode
-from frostfs_testlib.storage.grpc_operations import interfaces
-from frostfs_testlib.storage.grpc_operations.implementations.chunks import ChunksOperations
-from frostfs_testlib.testing.test_control import wait_for_success
-from frostfs_testlib.utils import cli_utils, file_utils
-
-logger = logging.getLogger("NeoLogger")
-
-
-class ObjectOperations(interfaces.ObjectInterface):
-    def __init__(self, cli: FrostfsCli) -> None:
-        self.cli = cli
-        self.chunks: interfaces.ChunksInterface = ChunksOperations(self.cli)
-
-    @reporter.step("Delete object")
-    def delete(
-        self,
-        cid: str,
-        oid: str,
-        endpoint: str,
-        bearer: str = "",
-        xhdr: Optional[dict] = None,
-        session: Optional[str] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> str:
-        """
-        DELETE an Object.
-
-        Args:
-            cid: ID of Container where we get the Object from
-            oid: ID of Object we are going to delete
-            bearer: path to Bearer Token file, appends to `--bearer` key
-            endpoint: FrostFS endpoint to send request to, appends to `--rpc-endpoint` key
-            xhdr: Request X-Headers in form of Key=Value
-            session: path to a JSON-encoded container session token
-            timeout: Timeout for the operation.
-        Returns:
-            (str): Tombstone ID
-        """
-        result = self.cli.object.delete(
-            rpc_endpoint=endpoint,
-            cid=cid,
-            oid=oid,
-            bearer=bearer,
-            xhdr=xhdr,
-            session=session,
-            timeout=timeout,
-        )
-
-        id_str = result.stdout.split("\n")[1]
-        tombstone = id_str.split(":")[1]
-        return tombstone.strip()
-
-    @reporter.step("Get object")
-    def get(
-        self,
-        cid: str,
-        oid: str,
-        endpoint: str,
-        bearer: Optional[str] = None,
-        write_object: Optional[str] = None,
-        xhdr: Optional[dict] = None,
-        no_progress: bool = True,
-        session: Optional[str] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> file_utils.TestFile:
-        """
-        GET from FrostFS.
-
-        Args:
-            cid (str): ID of Container where we get the Object from
-            oid (str): Object ID
-            bearer: path to Bearer Token file, appends to `--bearer` key
-            write_object: path to downloaded file, appends to `--file` key
-            endpoint: FrostFS endpoint to send request to, appends to `--rpc-endpoint` key
-            no_progress(optional, bool): do not show progress bar
-            xhdr (optional, dict): Request X-Headers in form of Key=Value
-            session (optional, dict): path to a JSON-encoded container session token
-            timeout: Timeout for the operation.
-        Returns:
-            (str): path to downloaded file
-        """
-        if not write_object:
-            write_object = str(uuid.uuid4())
-        test_file = file_utils.TestFile(os.path.join(ASSETS_DIR, write_object))
-
-        self.cli.object.get(
-            rpc_endpoint=endpoint,
-            cid=cid,
-            oid=oid,
-            file=test_file,
-            bearer=bearer,
-            no_progress=no_progress,
-            xhdr=xhdr,
-            session=session,
-            timeout=timeout,
-        )
-
-        return test_file
-
-    @reporter.step("Get object from random node")
-    def get_from_random_node(
-        self,
-        cid: str,
-        oid: str,
-        cluster: Cluster,
-        bearer: Optional[str] = None,
-        write_object: Optional[str] = None,
-        xhdr: Optional[dict] = None,
-        no_progress: bool = True,
-        session: Optional[str] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> str:
-        """
-        GET from FrostFS random storage node
-
-        Args:
-            cid: ID of Container where we get the Object from
-            oid: Object ID
-            cluster: cluster object
-            bearer (optional, str): path to Bearer Token file, appends to `--bearer` key
-            write_object (optional, str): path to downloaded file, appends to `--file` key
-            no_progress(optional, bool): do not show progress bar
-            xhdr (optional, dict): Request X-Headers in form of Key=Value
-            session (optional, dict): path to a JSON-encoded container session token
-            timeout: Timeout for the operation.
-        Returns:
-            (str): path to downloaded file
-        """
-        endpoint = cluster.get_random_storage_rpc_endpoint()
-        return self.get(
-            cid,
-            oid,
-            endpoint,
-            bearer,
-            write_object,
-            xhdr,
-            no_progress,
-            session,
-            timeout,
-        )
-
-    @reporter.step("Get hash object")
-    def hash(
-        self,
-        rpc_endpoint: str,
-        cid: str,
-        oid: str,
-        address: Optional[str] = None,
-        bearer: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        range: Optional[str] = None,
-        salt: Optional[str] = None,
-        ttl: Optional[int] = None,
-        session: Optional[str] = None,
-        hash_type: Optional[str] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> str:
-        """
-        Get object hash.
-
-        Args:
-            address: Address of wallet account.
-            bearer: File with signed JSON or binary encoded bearer token.
-            cid: Container ID.
-            generate_key: Generate new private key.
-            oid: Object ID.
-            range: Range to take hash from in the form offset1:length1,...
-            rpc_endpoint: Remote node address (as 'multiaddr' or '<host>:<port>').
-            salt: Salt in hex format.
-            ttl: TTL value in request meta header (default 2).
-            session: Filepath to a JSON- or binary-encoded token of the object RANGEHASH session.
-            hash_type: Hash type. Either 'sha256' or 'tz' (default "sha256").
-            wallet: WIF (NEP-2) string or path to the wallet or binary key.
-            xhdr: Dict with request X-Headers.
-            timeout: Timeout for the operation (default 15s).
-
-        Returns:
-            Command's result.
-        """
-        result = self.cli.object.hash(
-            rpc_endpoint=rpc_endpoint,
-            cid=cid,
-            oid=oid,
-            address=address,
-            bearer=bearer,
-            generate_key=generate_key,
-            range=range,
-            salt=salt,
-            ttl=ttl,
-            xhdr=xhdr,
-            session=session,
-            hash_type=hash_type,
-            timeout=timeout,
-        )
-        return result.stdout
-
-    @reporter.step("Head object")
-    def head(
-        self,
-        cid: str,
-        oid: str,
-        endpoint: str,
-        bearer: str = "",
-        xhdr: Optional[dict] = None,
-        json_output: bool = True,
-        is_raw: bool = False,
-        is_direct: bool = False,
-        session: Optional[str] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> CommandResult | Any:
-        """
-        HEAD an Object.
-
-        Args:
-            cid (str): ID of Container where we get the Object from
-            oid (str): ObjectID to HEAD
-            bearer (optional, str): path to Bearer Token file, appends to `--bearer` key
-            endpoint(optional, str): FrostFS endpoint to send request to
-            json_output(optional, bool): return response in JSON format or not; this flag
-                                        turns into `--json` key
-            is_raw(optional, bool): send "raw" request or not; this flag
-                                        turns into `--raw` key
-            is_direct(optional, bool): send request directly to the node or not; this flag
-                                        turns into `--ttl 1` key
-            xhdr (optional, dict): Request X-Headers in form of Key=Value
-            session (optional, dict): path to a JSON-encoded container session token
-            timeout: Timeout for the operation.
-        Returns:
-            depending on the `json_output` parameter value, the function returns
-            (dict): HEAD response in JSON format
-            or
-            (str): HEAD response as a plain text
-        """
-        result = self.cli.object.head(
-            rpc_endpoint=endpoint,
-            cid=cid,
-            oid=oid,
-            bearer=bearer,
-            json_mode=json_output,
-            raw=is_raw,
-            ttl=1 if is_direct else None,
-            xhdr=xhdr,
-            session=session,
-            timeout=timeout,
-        )
-
-        if not json_output:
-            return result
-
-        try:
-            decoded = json.loads(result.stdout)
-        except Exception as exc:
-            # If we failed to parse output as JSON, the cause might be
-            # the plain text string in the beginning of the output.
-            # Here we cut off first string and try to parse again.
-            logger.info(f"failed to parse output: {exc}")
-            logger.info("parsing output in another way")
-            fst_line_idx = result.stdout.find("\n")
-            decoded = json.loads(result.stdout[fst_line_idx:])
-
-        # if response
-        if "chunks" in decoded.keys():
-            logger.info("decoding ec chunks")
-            return decoded["chunks"]
-
-        # If response is Complex Object header, it has `splitId` key
-        if "splitId" in decoded.keys():
-            logger.info("decoding split header")
-            return utils.json_utils.decode_split_header(decoded)
-
-        # If response is Last or Linking Object header,
-        # it has `header` dictionary and non-null `split` dictionary
-        if "split" in decoded["header"].keys():
-            if decoded["header"]["split"]:
-                logger.info("decoding linking object")
-                return utils.json_utils.decode_linking_object(decoded)
-
-        if decoded["header"]["objectType"] == "STORAGE_GROUP":
-            logger.info("decoding storage group")
-            return utils.json_utils.decode_storage_group(decoded)
-
-        if decoded["header"]["objectType"] == "TOMBSTONE":
-            logger.info("decoding tombstone")
-            return utils.json_utils.decode_tombstone(decoded)
-
-        logger.info("decoding simple header")
-        return utils.json_utils.decode_simple_header(decoded)
-
-    @reporter.step("Lock Object")
-    def lock(
-        self,
-        cid: str,
-        oid: str,
-        endpoint: str,
-        lifetime: Optional[int] = None,
-        expire_at: Optional[int] = None,
-        address: Optional[str] = None,
-        bearer: Optional[str] = None,
-        session: Optional[str] = None,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> str:
-        """
-        Locks object in container.
-
-        Args:
-            address: Address of wallet account.
-            bearer: File with signed JSON or binary encoded bearer token.
-            cid: Container ID.
-            oid: Object ID.
-            lifetime: Lock lifetime.
-            expire_at: Lock expiration epoch.
-            shell: executor for cli command
-            endpoint: FrostFS endpoint to send request to, appends to `--rpc-endpoint` key
-            session: Path to a JSON-encoded container session token.
-            ttl: TTL value in request meta header (default 2).
-            wallet: WIF (NEP-2) string or path to the wallet or binary key.
-            xhdr: Dict with request X-Headers.
-            timeout: Timeout for the operation.
-
-        Returns:
-            Lock object ID
-        """
-        result = self.cli.object.lock(
-            rpc_endpoint=endpoint,
-            lifetime=lifetime,
-            expire_at=expire_at,
-            address=address,
-            cid=cid,
-            oid=oid,
-            bearer=bearer,
-            xhdr=xhdr,
-            session=session,
-            ttl=ttl,
-            timeout=timeout,
-        )
-
-        # Splitting CLI output to separate lines and taking the penultimate line
-        id_str = result.stdout.strip().split("\n")[0]
-        oid = id_str.split(":")[1]
-        return oid.strip()
-
-    @reporter.step("Put object")
-    def put(
-        self,
-        path: str,
-        cid: str,
-        endpoint: str,
-        bearer: Optional[str] = None,
-        copies_number: Optional[int] = None,
-        attributes: Optional[dict] = None,
-        xhdr: Optional[dict] = None,
-        expire_at: Optional[int] = None,
-        no_progress: bool = True,
-        session: Optional[str] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> str:
-        """
-        PUT of given file.
-
-        Args:
-            path: path to file to be PUT
-            cid: ID of Container where we get the Object from
-            bearer: path to Bearer Token file, appends to `--bearer` key
-            copies_number: Number of copies of the object to store within the RPC call
-            attributes: User attributes in form of Key1=Value1,Key2=Value2
-            endpoint: FrostFS endpoint to send request to, appends to `--rpc-endpoint` key
-            no_progress: do not show progress bar
-            expire_at: Last epoch in the life of the object
-            xhdr: Request X-Headers in form of Key=Value
-            session: path to a JSON-encoded container session token
-            timeout: Timeout for the operation.
-        Returns:
-            (str): ID of uploaded Object
-        """
-        result = self.cli.object.put(
-            rpc_endpoint=endpoint,
-            file=path,
-            cid=cid,
-            attributes=attributes,
-            bearer=bearer,
-            copies_number=copies_number,
-            expire_at=expire_at,
-            no_progress=no_progress,
-            xhdr=xhdr,
-            session=session,
-            timeout=timeout,
-        )
-
-        # Splitting CLI output to separate lines and taking the penultimate line
-        id_str = result.stdout.strip().split("\n")[-2]
-        oid = id_str.split(":")[1]
-        return oid.strip()
-
-    @reporter.step("Put object to random node")
-    def put_to_random_node(
-        self,
-        path: str,
-        cid: str,
-        cluster: Cluster,
-        bearer: Optional[str] = None,
-        copies_number: Optional[int] = None,
-        attributes: Optional[dict] = None,
-        xhdr: Optional[dict] = None,
-        expire_at: Optional[int] = None,
-        no_progress: bool = True,
-        session: Optional[str] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> str:
-        """
-        PUT of given file to a random storage node.
-
-        Args:
-            path: path to file to be PUT
-            cid: ID of Container where we get the Object from
-            cluster: cluster under test
-            bearer: path to Bearer Token file, appends to `--bearer` key
-            copies_number: Number of copies of the object to store within the RPC call
-            attributes: User attributes in form of Key1=Value1,Key2=Value2
-            cluster: cluster under test
-            no_progress: do not show progress bar
-            expire_at: Last epoch in the life of the object
-            xhdr: Request X-Headers in form of Key=Value
-            session: path to a JSON-encoded container session token
-            timeout: Timeout for the operation.
-        Returns:
-            ID of uploaded Object
-        """
-        endpoint = cluster.get_random_storage_rpc_endpoint()
-        return self.put(
-            path,
-            cid,
-            endpoint,
-            bearer,
-            copies_number,
-            attributes,
-            xhdr,
-            expire_at,
-            no_progress,
-            session,
-            timeout=timeout,
-        )
-
-    @reporter.step("Get Range")
-    def range(
-        self,
-        cid: str,
-        oid: str,
-        range_cut: str,
-        endpoint: str,
-        bearer: str = "",
-        xhdr: Optional[dict] = None,
-        session: Optional[str] = None,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> tuple[file_utils.TestFile, bytes]:
-        """
-        GETRANGE an Object.
-
-        Args:
-            wallet: wallet on whose behalf GETRANGE is done
-            cid: ID of Container where we get the Object from
-            oid: ID of Object we are going to request
-            range_cut: range to take data from in the form offset:length
-            shell: executor for cli command
-            endpoint: FrostFS endpoint to send request to, appends to `--rpc-endpoint` key
-            bearer: path to Bearer Token file, appends to `--bearer` key
-            xhdr: Request X-Headers in form of Key=Value
-            session: path to a JSON-encoded container session token
-            timeout: Timeout for the operation.
-        Returns:
-            (str, bytes) - path to the file with range content and content of this file as bytes
-        """
-        test_file = file_utils.TestFile(os.path.join(ASSETS_DIR, str(uuid.uuid4())))
-
-        self.cli.object.range(
-            rpc_endpoint=endpoint,
-            cid=cid,
-            oid=oid,
-            range=range_cut,
-            file=test_file,
-            bearer=bearer,
-            xhdr=xhdr,
-            session=session,
-            timeout=timeout,
-        )
-
-        with open(test_file, "rb") as file:
-            content = file.read()
-        return test_file, content
-
-    @reporter.step("Search object")
-    def search(
-        self,
-        cid: str,
-        endpoint: str,
-        bearer: str = "",
-        oid: Optional[str] = None,
-        filters: Optional[dict] = None,
-        expected_objects_list: Optional[list] = None,
-        xhdr: Optional[dict] = None,
-        session: Optional[str] = None,
-        phy: bool = False,
-        root: bool = False,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-        address: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        ttl: Optional[int] = None,
-    ) -> list:
-        """
-        SEARCH an Object.
-
-        Args:
-            wallet: wallet on whose behalf SEARCH is done
-            cid: ID of Container where we get the Object from
-            shell: executor for cli command
-            bearer: path to Bearer Token file, appends to `--bearer` key
-            endpoint: FrostFS endpoint to send request to, appends to `--rpc-endpoint` key
-            filters: key=value pairs to filter Objects
-            expected_objects_list: a list of ObjectIDs to compare found Objects with
-            xhdr: Request X-Headers in form of Key=Value
-            session: path to a JSON-encoded container session token
-            phy: Search physically stored objects.
-            root: Search for user objects.
-            timeout: Timeout for the operation.
-
-        Returns:
-            list of found ObjectIDs
-        """
-        result = self.cli.object.search(
-            rpc_endpoint=endpoint,
-            cid=cid,
-            bearer=bearer,
-            oid=oid,
-            xhdr=xhdr,
-            filters=[f"{filter_key} EQ {filter_val}" for filter_key, filter_val in filters.items()] if filters else None,
-            session=session,
-            phy=phy,
-            root=root,
-            address=address,
-            generate_key=generate_key,
-            ttl=ttl,
-            timeout=timeout,
-        )
-
-        found_objects = re.findall(r"(\w{43,44})", result.stdout)
-
-        if expected_objects_list:
-            if sorted(found_objects) == sorted(expected_objects_list):
-                logger.info(f"Found objects list '{found_objects}' " f"is equal for expected list '{expected_objects_list}'")
-            else:
-                logger.warning(f"Found object list {found_objects} " f"is not equal to expected list '{expected_objects_list}'")
-
-        return found_objects
-
-    @wait_for_success()
-    @reporter.step("Search object nodes")
-    def nodes(
-        self,
-        cluster: Cluster,
-        cid: str,
-        oid: str,
-        alive_node: ClusterNode,
-        bearer: str = "",
-        xhdr: Optional[dict] = None,
-        is_direct: bool = False,
-        verify_presence_all: bool = False,
-        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> list[ClusterNode]:
-        endpoint = alive_node.storage_node.get_rpc_endpoint()
-
-        response = self.cli.object.nodes(
-            rpc_endpoint=endpoint,
-            cid=cid,
-            oid=oid,
-            bearer=bearer,
-            ttl=1 if is_direct else None,
-            json=True,
-            xhdr=xhdr,
-            timeout=timeout,
-            verify_presence_all=verify_presence_all,
-        )
-
-        response_json = json.loads(response.stdout)
-        # Currently, the command will show expected and confirmed nodes.
-        # And we (currently) count only nodes which are both expected and confirmed
-        object_nodes_id = {
-            required_node
-            for data_object in response_json["data_objects"]
-            for required_node in data_object["required_nodes"]
-            if required_node in data_object["confirmed_nodes"]
-        }
-
-        netmap_nodes_list = cli_utils.parse_netmap_output(
-            self.cli.netmap.snapshot(
-                rpc_endpoint=endpoint,
-            ).stdout
-        )
-        netmap_nodes = [
-            netmap_node for object_node in object_nodes_id for netmap_node in netmap_nodes_list if object_node == netmap_node.node_id
-        ]
-
-        object_nodes = [
-            cluster_node
-            for netmap_node in netmap_nodes
-            for cluster_node in cluster.cluster_nodes
-            if netmap_node.node == cluster_node.host_ip
-        ]
-
-        return object_nodes

src/frostfs_testlib/storage/grpc_operations/interfaces.py

@@ -1,392 +0,0 @@
-from abc import ABC, abstractmethod
-from typing import Any, List, Optional
-
-from frostfs_testlib.shell.interfaces import CommandResult
-from frostfs_testlib.storage.cluster import Cluster, ClusterNode
-from frostfs_testlib.storage.constants import PlacementRule
-from frostfs_testlib.storage.dataclasses.storage_object_info import Chunk, NodeNetmapInfo
-from frostfs_testlib.utils import file_utils
-
-
-class ChunksInterface(ABC):
-    @abstractmethod
-    def search_node_without_chunks(self, chunks: list[Chunk], cluster: Cluster, endpoint: str = None) -> list[ClusterNode]:
-        pass
-
-    @abstractmethod
-    def get_chunk_node(self, cluster: Cluster, chunk: Chunk) -> tuple[ClusterNode, NodeNetmapInfo]:
-        pass
-
-    @abstractmethod
-    def get_shard_chunk(self, node: ClusterNode, chunk: Chunk) -> str:
-        pass
-
-    @abstractmethod
-    def get_all(
-        self,
-        rpc_endpoint: str,
-        cid: str,
-        oid: str,
-        wallet: Optional[str] = None,
-        address: Optional[str] = None,
-        bearer: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        trace: bool = False,
-        root: bool = False,
-        verify_presence_all: bool = False,
-        json: bool = True,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = None,
-    ) -> list[Chunk]:
-        pass
-
-    @abstractmethod
-    def get_parity(
-        self,
-        rpc_endpoint: str,
-        cid: str,
-        wallet: Optional[str] = None,
-        address: Optional[str] = None,
-        bearer: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        oid: Optional[str] = None,
-        trace: bool = False,
-        root: bool = False,
-        verify_presence_all: bool = False,
-        json: bool = True,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = None,
-    ) -> Chunk:
-        pass
-
-    @abstractmethod
-    def get_first_data(
-        self,
-        rpc_endpoint: str,
-        cid: str,
-        wallet: Optional[str] = None,
-        address: Optional[str] = None,
-        bearer: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        oid: Optional[str] = None,
-        trace: bool = False,
-        root: bool = False,
-        verify_presence_all: bool = False,
-        json: bool = True,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = None,
-    ) -> Chunk:
-        pass
-
-
-class ObjectInterface(ABC):
-    def __init__(self) -> None:
-        self.chunks: ChunksInterface
-
-    @abstractmethod
-    def delete(
-        self,
-        cid: str,
-        oid: str,
-        endpoint: str,
-        bearer: str = "",
-        xhdr: Optional[dict] = None,
-        session: Optional[str] = None,
-        timeout: Optional[str] = None,
-    ) -> str:
-        pass
-
-    @abstractmethod
-    def get(
-        self,
-        cid: str,
-        oid: str,
-        endpoint: str,
-        bearer: Optional[str] = None,
-        write_object: Optional[str] = None,
-        xhdr: Optional[dict] = None,
-        no_progress: bool = True,
-        session: Optional[str] = None,
-        timeout: Optional[str] = None,
-    ) -> file_utils.TestFile:
-        pass
-
-    @abstractmethod
-    def get_from_random_node(
-        self,
-        cid: str,
-        oid: str,
-        cluster: Cluster,
-        bearer: Optional[str] = None,
-        write_object: Optional[str] = None,
-        xhdr: Optional[dict] = None,
-        no_progress: bool = True,
-        session: Optional[str] = None,
-        timeout: Optional[str] = None,
-    ) -> str:
-        pass
-
-    @abstractmethod
-    def hash(
-        self,
-        endpoint: str,
-        cid: str,
-        oid: str,
-        address: Optional[str] = None,
-        bearer: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        range: Optional[str] = None,
-        salt: Optional[str] = None,
-        ttl: Optional[int] = None,
-        session: Optional[str] = None,
-        hash_type: Optional[str] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = None,
-    ) -> str:
-        pass
-
-    @abstractmethod
-    def head(
-        self,
-        cid: str,
-        oid: str,
-        endpoint: str,
-        bearer: str = "",
-        xhdr: Optional[dict] = None,
-        json_output: bool = True,
-        is_raw: bool = False,
-        is_direct: bool = False,
-        session: Optional[str] = None,
-        timeout: Optional[str] = None,
-    ) -> CommandResult | Any:
-        pass
-
-    @abstractmethod
-    def lock(
-        self,
-        cid: str,
-        oid: str,
-        endpoint: str,
-        lifetime: Optional[int] = None,
-        expire_at: Optional[int] = None,
-        address: Optional[str] = None,
-        bearer: Optional[str] = None,
-        session: Optional[str] = None,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = None,
-    ) -> str:
-        pass
-
-    @abstractmethod
-    def put(
-        self,
-        path: str,
-        cid: str,
-        endpoint: str,
-        bearer: Optional[str] = None,
-        copies_number: Optional[int] = None,
-        attributes: Optional[dict] = None,
-        xhdr: Optional[dict] = None,
-        expire_at: Optional[int] = None,
-        no_progress: bool = True,
-        session: Optional[str] = None,
-        timeout: Optional[str] = None,
-    ) -> str:
-        pass
-
-    @abstractmethod
-    def put_to_random_node(
-        self,
-        path: str,
-        cid: str,
-        cluster: Cluster,
-        bearer: Optional[str] = None,
-        copies_number: Optional[int] = None,
-        attributes: Optional[dict] = None,
-        xhdr: Optional[dict] = None,
-        expire_at: Optional[int] = None,
-        no_progress: bool = True,
-        session: Optional[str] = None,
-        timeout: Optional[str] = None,
-    ) -> str:
-        pass
-
-    @abstractmethod
-    def range(
-        self,
-        cid: str,
-        oid: str,
-        range_cut: str,
-        endpoint: str,
-        bearer: str = "",
-        xhdr: Optional[dict] = None,
-        session: Optional[str] = None,
-        timeout: Optional[str] = None,
-    ) -> tuple[file_utils.TestFile, bytes]:
-        pass
-
-    @abstractmethod
-    def search(
-        self,
-        cid: str,
-        endpoint: str,
-        bearer: str = "",
-        oid: Optional[str] = None,
-        filters: Optional[dict] = None,
-        expected_objects_list: Optional[list] = None,
-        xhdr: Optional[dict] = None,
-        session: Optional[str] = None,
-        phy: bool = False,
-        root: bool = False,
-        timeout: Optional[str] = None,
-        address: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        ttl: Optional[int] = None,
-    ) -> List:
-        pass
-
-    @abstractmethod
-    def nodes(
-        self,
-        cluster: Cluster,
-        cid: str,
-        oid: str,
-        alive_node: ClusterNode,
-        bearer: str = "",
-        xhdr: Optional[dict] = None,
-        is_direct: bool = False,
-        verify_presence_all: bool = False,
-        timeout: Optional[str] = None,
-    ) -> List[ClusterNode]:
-        pass
-
-
-class ContainerInterface(ABC):
-    @abstractmethod
-    def create(
-        self,
-        endpoint: str,
-        nns_zone: Optional[str] = None,
-        nns_name: Optional[str] = None,
-        address: Optional[str] = None,
-        attributes: Optional[dict] = None,
-        basic_acl: Optional[str] = None,
-        await_mode: bool = False,
-        disable_timestamp: bool = False,
-        force: bool = False,
-        trace: bool = False,
-        name: Optional[str] = None,
-        nonce: Optional[str] = None,
-        policy: Optional[str] = None,
-        session: Optional[str] = None,
-        subnet: Optional[str] = None,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = None,
-    ) -> str:
-        """
-        Create a new container and register it in the FrostFS.
-        It will be stored in the sidechain when the Inner Ring accepts it.
-        """
-        raise NotImplementedError("No implemethed method create")
-
-    @abstractmethod
-    def delete(
-        self,
-        endpoint: str,
-        cid: str,
-        address: Optional[str] = None,
-        await_mode: bool = False,
-        session: Optional[str] = None,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        force: bool = False,
-        trace: bool = False,
-    ) -> List[str]:
-        """
-        Delete an existing container.
-        Only the owner of the container has permission to remove the container.
-        """
-        raise NotImplementedError("No implemethed method delete")
-
-    @abstractmethod
-    def get(
-        self,
-        endpoint: str,
-        cid: str,
-        address: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        await_mode: bool = False,
-        to: Optional[str] = None,
-        json_mode: bool = True,
-        trace: bool = False,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = None,
-    ) -> List[str]:
-        """Get container field info."""
-        raise NotImplementedError("No implemethed method get")
-
-    @abstractmethod
-    def get_eacl(
-        self,
-        endpoint: str,
-        cid: str,
-        address: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        await_mode: bool = False,
-        json_mode: bool = True,
-        trace: bool = False,
-        to: Optional[str] = None,
-        session: Optional[str] = None,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = None,
-    ) -> List[str]:
-        """Get extended ACL table of container."""
-        raise NotImplementedError("No implemethed method get-eacl")
-
-    @abstractmethod
-    def list(
-        self,
-        endpoint: str,
-        name: Optional[str] = None,
-        address: Optional[str] = None,
-        generate_key: Optional[bool] = None,
-        trace: bool = False,
-        owner: Optional[str] = None,
-        ttl: Optional[int] = None,
-        xhdr: Optional[dict] = None,
-        timeout: Optional[str] = None,
-        **params,
-    ) -> List[str]:
-        """List all created containers."""
-        raise NotImplementedError("No implemethed method list")
-
-    @abstractmethod
-    def nodes(
-        self,
-        endpoint: str,
-        cid: str,
-        cluster: Cluster,
-        address: Optional[str] = None,
-        ttl: Optional[int] = None,
-        from_file: Optional[str] = None,
-        trace: bool = False,
-        short: Optional[bool] = True,
-        xhdr: Optional[dict] = None,
-        generate_key: Optional[bool] = None,
-        timeout: Optional[str] = None,
-    ) -> List[ClusterNode]:
-        """Show the nodes participating in the container in the current epoch."""
-        raise NotImplementedError("No implemethed method nodes")
-
-
-class GrpcClientWrapper(ABC):
-    def __init__(self) -> None:
-        self.object: ObjectInterface
-        self.container: ContainerInterface