Test-help

Signed-off-by: a.berezin <a.berezin@yadro.com>

.devenv.hosting.yaml

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

README.md

@@ -1,4 +1,4 @@
-# frostfs-testlib
+123123# frostfs-testlib
 This library provides building blocks and utilities to facilitate development of automated tests for FrostFS system.
 
 ## Installation

src/frostfs_testlib/cli/frostfs_cli/shards.py

@@ -143,3 +143,101 @@ class FrostfsCliShards(CliCommand):
             **{param: value for param, value in locals().items() if param not in ["self", "wallet_password"]},
         )
 
+    def evacuation_start(
+        self,
+        endpoint: str,
+        id: Optional[str] = None,
+        scope: Optional[str] = None,
+        all: bool = False,
+        no_errors: bool = True,
+        await_mode: bool = False,
+        address: Optional[str] = None,
+        timeout: Optional[str] = None,
+        no_progress: bool = False,
+    ) -> CommandResult:
+        """
+        Objects evacuation from shard to other shards.
+
+        Args:
+            address: Address of wallet account
+            all: Process all shards
+            await: Block execution until evacuation is completed
+            endpoint: Remote node control address (as 'multiaddr' or '<host>:<port>')
+            id: List of shard IDs in base58 encoding
+            no_errors: Skip invalid/unreadable objects (default true)
+            no_progress: Print progress if await provided
+            scope: Evacuation scope; possible values: trees, objects, all (default "all")
+            timeout: Timeout for an operation (default 15s)
+
+        Returns:
+            Command's result.
+        """
+        return self._execute(
+            "control shards evacuation start",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )
+
+    def evacuation_reset(
+        self,
+        endpoint: str,
+        address: Optional[str] = None,
+        timeout: Optional[str] = None,
+    ) -> CommandResult:
+        """
+        Reset evacuate objects from shard to other shards status.
+
+        Args:
+            address: Address of wallet account
+            endpoint: Remote node control address (as 'multiaddr' or '<host>:<port>')
+            timeout: Timeout for an operation (default 15s)
+        Returns:
+            Command's result.
+        """
+        return self._execute(
+            "control shards evacuation reset",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )
+
+    def evacuation_stop(
+        self,
+        endpoint: str,
+        address: Optional[str] = None,
+        timeout: Optional[str] = None,
+    ) -> CommandResult:
+        """
+        Stop running evacuate process from shard to other shards.
+
+        Args:
+            address: Address of wallet account
+            endpoint: Remote node control address (as 'multiaddr' or '<host>:<port>')
+            timeout: Timeout for an operation (default 15s)
+
+        Returns:
+            Command's result.
+        """
+        return self._execute(
+            "control shards evacuation stop",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )
+
+    def evacuation_status(
+        self,
+        endpoint: str,
+        address: Optional[str] = None,
+        timeout: Optional[str] = None,
+    ) -> CommandResult:
+        """
+        Get evacuate objects from shard to other shards status.
+
+        Args:
+            address: Address of wallet account
+            endpoint: Remote node control address (as 'multiaddr' or '<host>:<port>')
+            timeout: Timeout for an operation (default 15s)
+
+        Returns:
+            Command's result.
+        """
+        return self._execute(
+            "control shards evacuation status",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )

src/frostfs_testlib/hosting/config.py

@@ -60,6 +60,7 @@ class HostConfig:
     """
 
     plugin_name: str
+    hostname: str
     healthcheck_plugin_name: str
     address: str
     s3_creds_plugin_name: str = field(default="authmate")

src/frostfs_testlib/resources/common.py

@@ -51,3 +51,5 @@ 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/s3/aws_cli_client.py

@@ -975,7 +975,7 @@ class AwsCliClient(S3ClientWrapper):
         response = self._to_json(output)
 
         assert response.get("Policy"), f"Expected Policy in response:\n{response}"
-        assert response["Policy"].get("PolicyName") == policy_name, f"PolicyName should be equal to {policy_name}"
+        assert response["Policy"].get("Arn") == policy_arn, f"PolicyArn should be equal to {policy_arn}"
 
         return response
 

src/frostfs_testlib/s3/boto3_client.py

@@ -776,7 +776,7 @@ class Boto3ClientWrapper(S3ClientWrapper):
     def iam_get_policy(self, policy_arn: str) -> dict:
         response = self.boto3_iam_client.get_policy(PolicyArn=policy_arn)
         assert response.get("Policy"), f"Expected Policy in response:\n{response}"
-        assert response["Policy"].get("PolicyName") == policy_name, f"PolicyName should be equal to {policy_name}"
+        assert response["Policy"].get("Arn") == policy_arn, f"PolicyArn should be equal to {policy_arn}"
 
         return response
 

src/frostfs_testlib/s3/interfaces.py

@@ -408,7 +408,7 @@ class S3ClientWrapper(HumanReadableABC):
         """Adds the specified user to the specified group"""
 
     @abstractmethod
-    def iam_attach_group_policy(self, group: str, policy_arn: str) -> dict:
+    def iam_attach_group_policy(self, group_name: str, policy_arn: str) -> dict:
         """Attaches the specified managed policy to the specified IAM group"""
 
     @abstractmethod

src/frostfs_testlib/shell/local_shell.py

@@ -1,15 +1,18 @@
 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):
@@ -28,7 +31,7 @@ class LocalShell(Shell):
         for inspector in [*self.command_inspectors, *extra_inspectors]:
             command = inspector.inspect(original_command, command)
 
-        with reporter.step(f"Executing command: {command}"):
+        with step_context(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/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_cmd_table, parse_netmap_output
+from frostfs_testlib.utils.cli_utils import parse_netmap_output
 from frostfs_testlib.utils.file_utils import TestFile
 
 logger = logging.getLogger("NeoLogger")
@@ -623,25 +623,20 @@ 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)
 
 
@@ -695,11 +690,13 @@ 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]
-    validated_state = second_line.split(":")
-    return {
-        latest_block[0].replace(":", ""): int(latest_block[1]),
-        validated_state[0].replace(":", ""): int(validated_state[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/storage/controllers/cluster_state_controller.py

@@ -538,4 +538,4 @@ class ClusterStateController:
             shell=cluster_node.host.get_shell(),
             frostfs_adm_exec_path=FROSTFS_ADM_EXEC,
         )
-        return frostfs_adm.morph.dump_hashes(cluster_node.morph_chain.get_endpoint(), domain_name).stdout
+        return frostfs_adm.morph.dump_hashes(cluster_node.morph_chain.get_http_endpoint(), domain_name).stdout

src/frostfs_testlib/storage/controllers/shards_watcher.py

@@ -2,22 +2,22 @@ import json
 from typing import Any
 
 from frostfs_testlib.cli.frostfs_cli.shards import FrostfsCliShards
+from frostfs_testlib.shell.interfaces import CommandResult
 from frostfs_testlib.storage.cluster import ClusterNode
 from frostfs_testlib.testing.test_control import wait_for_success
 
 
 class ShardsWatcher:
-    shards_snapshots: list[dict[str, Any]] = []
-
     def __init__(self, node_under_test: ClusterNode) -> None:
+        self.shards_snapshots: list[dict[str, Any]] = []
         self.storage_node = node_under_test.storage_node
         self.take_shards_snapshot()
 
-    def take_shards_snapshot(self):
+    def take_shards_snapshot(self) -> None:
         snapshot = self.get_shards_snapshot()
         self.shards_snapshots.append(snapshot)
 
-    def get_shards_snapshot(self):
+    def get_shards_snapshot(self) -> dict[str, Any]:
         shards_snapshot: dict[str, Any] = {}
 
         shards = self.get_shards()
@@ -26,17 +26,17 @@ class ShardsWatcher:
 
         return shards_snapshot
 
-    def _get_current_snapshot(self):
+    def _get_current_snapshot(self) -> dict[str, Any]:
         return self.shards_snapshots[-1]
 
-    def _get_previous_snapshot(self):
+    def _get_previous_snapshot(self) -> dict[str, Any]:
         return self.shards_snapshots[-2]
 
-    def _is_shard_present(self, shard_id):
+    def _is_shard_present(self, shard_id) -> bool:
         snapshot = self._get_current_snapshot()
         return shard_id in snapshot
 
-    def get_shards_with_new_errors(self):
+    def get_shards_with_new_errors(self) -> dict[str, Any]:
         current_snapshot = self._get_current_snapshot()
         previous_snapshot = self._get_previous_snapshot()
         shards_with_new_errors: dict[str, Any] = {}
@@ -46,7 +46,7 @@ class ShardsWatcher:
 
         return shards_with_new_errors
 
-    def get_shards_with_errors(self):
+    def get_shards_with_errors(self) -> dict[str, Any]:
         snapshot = self.get_shards_snapshot()
         shards_with_errors: dict[str, Any] = {}
         for shard_id, shard in snapshot.items():
@@ -55,7 +55,7 @@ class ShardsWatcher:
 
         return shards_with_errors
 
-    def get_shard_status(self, shard_id: str):
+    def get_shard_status(self, shard_id: str):  # -> Any:
         snapshot = self.get_shards_snapshot()
 
         assert shard_id in snapshot, f"Shard {shard_id} is missing: {snapshot}"
@@ -63,18 +63,18 @@ class ShardsWatcher:
         return snapshot[shard_id]["mode"]
 
     @wait_for_success(60, 2)
-    def await_for_all_shards_status(self, status: str):
+    def await_for_all_shards_status(self, status: str) -> None:
         snapshot = self.get_shards_snapshot()
 
         for shard_id in snapshot:
             assert snapshot[shard_id]["mode"] == status, f"Shard {shard_id} have wrong shard status"
 
     @wait_for_success(60, 2)
-    def await_for_shard_status(self, shard_id: str, status: str):
+    def await_for_shard_status(self, shard_id: str, status: str) -> None:
         assert self.get_shard_status(shard_id) == status
 
     @wait_for_success(60, 2)
-    def await_for_shard_have_new_errors(self, shard_id: str):
+    def await_for_shard_have_new_errors(self, shard_id: str) -> None:
         self.take_shards_snapshot()
         assert self._is_shard_present(shard_id)
         shards_with_new_errors = self.get_shards_with_new_errors()
@@ -82,7 +82,7 @@ class ShardsWatcher:
         assert shard_id in shards_with_new_errors, f"Expected shard {shard_id} to have new errors, but haven't {self.shards_snapshots[-1]}"
 
     @wait_for_success(300, 5)
-    def await_for_shards_have_no_new_errors(self):
+    def await_for_shards_have_no_new_errors(self) -> None:
         self.take_shards_snapshot()
         shards_with_new_errors = self.get_shards_with_new_errors()
         assert len(shards_with_new_errors) == 0
@@ -102,7 +102,7 @@ class ShardsWatcher:
 
         return json.loads(response.stdout.split(">", 1)[1])
 
-    def set_shard_mode(self, shard_id: str, mode: str, clear_errors: bool = True):
+    def set_shard_mode(self, shard_id: str, mode: str, clear_errors: bool = True) -> CommandResult:
         shards_cli = FrostfsCliShards(
             self.storage_node.host.get_shell(),
             self.storage_node.host.get_cli_config("frostfs-cli").exec_path,

src/frostfs_testlib/storage/dataclasses/ape.py

@@ -47,6 +47,8 @@ class ConditionType(HumanReadableEnum):
 class ConditionKey(HumanReadableEnum):
     ROLE = '"\\$Actor:role"'
     PUBLIC_KEY = '"\\$Actor:publicKey"'
+    OBJECT_TYPE = '"\\$Object:objectType"'
+    OBJECT_ID = '"\\$Object:objectID"'
 
 
 class MatchType(HumanReadableEnum):
@@ -75,6 +77,14 @@ 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/testing/parallel.py

@@ -1,10 +1,22 @@
 import itertools
 from concurrent.futures import Future, ThreadPoolExecutor
+from contextlib import contextmanager
 from typing import Callable, Collection, Optional, Union
 
 MAX_WORKERS = 50
 
 
+@contextmanager
+def parallel_workers_limit(workers_count: int):
+    global MAX_WORKERS
+    original_value = MAX_WORKERS
+    MAX_WORKERS = workers_count
+    try:
+        yield
+    finally:
+        MAX_WORKERS = original_value
+
+
 def parallel(
     fn: Union[Callable, list[Callable]],
     parallel_items: Optional[Collection] = None,