[#372] Added decorator wait until stabilization metric values

Signed-off-by: Ilyas Niyazov <i.niyazov@yadro.com>

pyproject.toml

@@ -28,7 +28,7 @@ dependencies = [
     "pytest==7.1.2",
     "tenacity==8.0.1",
     "boto3==1.35.30",
-    "boto3-stubs[essential]==1.35.30",
+    "boto3-stubs[s3,iam,sts]==1.35.30",
 ]
 requires-python = ">=3.10"
 

requirements.txt

@@ -9,7 +9,7 @@ testrail-api==1.12.0
 tenacity==8.0.1
 pytest==7.1.2
 boto3==1.35.30
-boto3-stubs[essential]==1.35.30
+boto3-stubs[s3,iam,sts]==1.35.30
 pydantic==2.10.6
 
 # Dev dependencies
@@ -22,4 +22,4 @@ pylint==2.17.4
 # Packaging dependencies
 build==0.8.0
 setuptools==65.3.0
-twine==4.0.1
+twine==4.0.1

src/frostfs_testlib/cli/netmap_parser.py

@@ -89,7 +89,7 @@ class NetmapParser:
                     return snapshot
 
     @staticmethod
-    def node_info(output: dict) -> NodeNetmapInfo:
+    def node_info(output: dict) -> NodeInfo:
         data_dict = {"attributes": {}}
 
         for key, value in output.items():

src/frostfs_testlib/clients/http/http_client.py

@@ -15,14 +15,14 @@ LOGGING_CONFIG = {
     "handlers": {"default": {"class": "logging.StreamHandler", "formatter": "http", "stream": "ext://sys.stderr"}},
     "formatters": {
         "http": {
-            "format": "%(levelname)s [%(asctime)s] %(name)s - %(message)s",
+            "format": "%(asctime)s [%(levelname)s] %(name)s - %(message)s",
             "datefmt": "%Y-%m-%d %H:%M:%S",
         }
     },
     "loggers": {
         "httpx": {
             "handlers": ["default"],
-            "level": "DEBUG",
+            "level": "ERROR",
         },
         "httpcore": {
             "handlers": ["default"],
@@ -43,7 +43,7 @@ class HttpClient:
         response = client.request(method, url, **kwargs)
 
         self._attach_response(response, **kwargs)
-        logger.info(f"Response: {response.status_code} => {response.text}")
+        # logger.info(f"Response: {response.status_code} => {response.text}")
 
         if expected_status_code:
             assert (
@@ -131,6 +131,7 @@ class HttpClient:
 
         reporter.attach(report, "Requests Info")
         reporter.attach(curl_request, "CURL")
+        cls._write_log(curl_request, response_body, response.status_code)
 
     @classmethod
     def _create_curl_request(cls, url: str, method: str, headers: httpx.Headers, data: str, files: dict) -> str:
@@ -143,3 +144,9 @@ class HttpClient:
 
         # Option -k means no verify SSL
         return f"curl {url} -X {method} {headers}{data} -k"
+
+    @classmethod
+    def _write_log(cls, curl: str, res_body: str, res_code: int) -> None:
+        if res_body:
+            curl += f"\nResponse: {res_code}\n{res_body}"
+        logger.info(f"{curl}")

src/frostfs_testlib/clients/s3/aws_cli_client.py

@@ -959,6 +959,15 @@ class AwsCliClient(S3ClientWrapper):
 
         return json_output
 
+    @reporter.step("Create presign url for the object")
+    def create_presign_url(self, method: str, bucket: str, key: str, expires_in: Optional[int] = 3600) -> str:
+        # AWS CLI does not support method definition and world only in 'get_object' state by default
+        cmd = f"aws {self.common_flags} s3 presign s3://{bucket}/{key} " f"--endpoint-url {self.s3gate_endpoint} --profile {self.profile}"
+        if expires_in:
+            cmd += f" --expires-in {expires_in}"
+        response = self.local_shell.exec(cmd).stdout
+        return response.strip()
+
     # IAM METHODS #
     # Some methods don't have checks because AWS is silent in some cases (delete, attach, etc.)
 

src/frostfs_testlib/clients/s3/boto3_client.py

@@ -10,7 +10,9 @@ import boto3
 import urllib3
 from botocore.config import Config
 from botocore.exceptions import ClientError
+from mypy_boto3_iam import IAMClient
 from mypy_boto3_s3 import S3Client
+from mypy_boto3_sts import STSClient
 
 from frostfs_testlib import reporter
 from frostfs_testlib.clients.s3.interfaces import S3ClientWrapper, VersioningStatus, _make_objs_dict
@@ -39,8 +41,8 @@ class Boto3ClientWrapper(S3ClientWrapper):
         self.boto3_client: S3Client = None
 
         self.iam_endpoint: str = ""
-        self.boto3_iam_client: S3Client = None
-        self.boto3_sts_client: S3Client = None
+        self.boto3_iam_client: IAMClient = None
+        self.boto3_sts_client: STSClient = None
 
         self.access_key_id = access_key_id
         self.secret_access_key = secret_access_key
@@ -48,7 +50,13 @@ class Boto3ClientWrapper(S3ClientWrapper):
         self.region = region
 
         self.session = boto3.Session()
-        self.config = Config(retries={"max_attempts": MAX_REQUEST_ATTEMPTS, "mode": RETRY_MODE})
+        self.config = Config(
+            signature_version="s3v4",
+            retries={
+                "max_attempts": MAX_REQUEST_ATTEMPTS,
+                "mode": RETRY_MODE,
+            },
+        )
 
         self.set_endpoint(s3gate_endpoint)
 
@@ -90,6 +98,7 @@ class Boto3ClientWrapper(S3ClientWrapper):
             aws_access_key_id=self.access_key_id,
             aws_secret_access_key=self.secret_access_key,
             endpoint_url=iam_endpoint,
+            region_name=self.region,
             verify=False,
         )
 
@@ -769,7 +778,7 @@ class Boto3ClientWrapper(S3ClientWrapper):
         return response.get("TagSet")
 
     @reporter.step("Delete object tagging")
-    def delete_object_tagging(self, bucket: str, key: str) -> None:
+    def delete_object_tagging(self, bucket: str, key: str, version_id: Optional[str] = None) -> None:
         params = self._convert_to_s3_params(locals())
         self._exec_request(
             self.boto3_client.delete_object_tagging,
@@ -812,6 +821,16 @@ class Boto3ClientWrapper(S3ClientWrapper):
     ) -> dict:
         raise NotImplementedError("Cp is not supported for boto3 client")
 
+    @reporter.step("Create presign url for the object")
+    def create_presign_url(self, method: str, bucket: str, key: str, expires_in: Optional[int] = 3600) -> str:
+        response = self._exec_request(
+            method=self.boto3_client.generate_presigned_url,
+            params={"ClientMethod": method, "Params": {"Bucket": bucket, "Key": key}, "ExpiresIn": expires_in},
+            endpoint=self.s3gate_endpoint,
+            profile=self.profile,
+        )
+        return response
+
     # END OBJECT METHODS #
 
     # IAM METHODS #

src/frostfs_testlib/clients/s3/interfaces.py

@@ -377,7 +377,7 @@ class S3ClientWrapper(HumanReadableABC):
         """Returns the tag-set of an object."""
 
     @abstractmethod
-    def delete_object_tagging(self, bucket: str, key: str) -> None:
+    def delete_object_tagging(self, bucket: str, key: str, version_id: Optional[str] = None) -> None:
         """Removes the entire tag set from the specified object."""
 
     @abstractmethod
@@ -425,6 +425,10 @@ class S3ClientWrapper(HumanReadableABC):
     ) -> dict:
         """cp directory TODO: Add proper description"""
 
+    @abstractmethod
+    def create_presign_url(self, method: str, bucket: str, key: str, expires_in: Optional[int] = 3600) -> str:
+        """Creates presign URL"""
+
     # END OF OBJECT METHODS #
 
     # IAM METHODS #

src/frostfs_testlib/hooks.py

@@ -16,6 +16,9 @@ def pytest_add_frostfs_marker(items: list[pytest.Item]):
 # pytest hook. Do not rename
 @pytest.hookimpl(trylast=True)
 def pytest_collection_modifyitems(items: list[pytest.Item]):
+    # The order of running tests corresponded to the suites
+    items.sort(key=lambda item: item.location[0])
+
     # Change order of tests based on @pytest.mark.order(<int>) marker
     def order(item: pytest.Item) -> int:
         order_marker = item.get_closest_marker("order")

src/frostfs_testlib/shell/local_shell.py

@@ -141,6 +141,6 @@ class LocalShell(Shell):
             f"RETCODE: {result.return_code}\n\n"
             f"STDOUT:\n{result.stdout}\n"
             f"STDERR:\n{result.stderr}\n"
-            f"Start / End / Elapsed\t {start_time.time()} / {end_time.time()} / {elapsed_time}"
+            f"Start / End / Elapsed\t {start_time} / {end_time} / {elapsed_time}"
         )
         reporter.attach(command_attachment, "Command execution.txt")

src/frostfs_testlib/shell/ssh_shell.py

@@ -68,8 +68,7 @@ class SshConnectionProvider:
             try:
                 if creds.ssh_key_path:
                     logger.info(
-                        f"Trying to connect to host {host} as {creds.ssh_login} using SSH key "
-                        f"{creds.ssh_key_path} (attempt {attempt})"
+                        f"Trying to connect to host {host} as {creds.ssh_login} using SSH key " f"{creds.ssh_key_path} (attempt {attempt})"
                     )
                     connection.connect(
                         hostname=host,
@@ -79,9 +78,7 @@ class SshConnectionProvider:
                         timeout=self.CONNECTION_TIMEOUT,
                     )
                 else:
-                    logger.info(
-                        f"Trying to connect to host {host} as {creds.ssh_login} using password " f"(attempt {attempt})"
-                    )
+                    logger.info(f"Trying to connect to host {host} as {creds.ssh_login} using password " f"(attempt {attempt})")
                     connection.connect(
                         hostname=host,
                         port=port,
@@ -104,9 +101,7 @@ class SshConnectionProvider:
                 connection.close()
                 can_retry = attempt + 1 < self.SSH_CONNECTION_ATTEMPTS
                 if can_retry:
-                    logger.warn(
-                        f"Can't connect to host {host}, will retry after {self.SSH_ATTEMPTS_INTERVAL}s. Error: {exc}"
-                    )
+                    logger.warn(f"Can't connect to host {host}, will retry after {self.SSH_ATTEMPTS_INTERVAL}s. Error: {exc}")
                     sleep(self.SSH_ATTEMPTS_INTERVAL)
                     continue
                 logger.exception(f"Can't connect to host {host}")
@@ -139,7 +134,7 @@ def log_command(func):
                 f"RC:\n {result.return_code}\n"
                 f"STDOUT:\n{textwrap.indent(result.stdout, ' ')}\n"
                 f"STDERR:\n{textwrap.indent(result.stderr, ' ')}\n"
-                f"Start / End / Elapsed\t {start_time.time()} / {end_time.time()} / {elapsed_time}"
+                f"Start / End / Elapsed\t {start_time} / {end_time} / {elapsed_time}"
             )
 
             if not options.no_log:
@@ -185,13 +180,11 @@ class SSHShell(Shell):
         private_key_passphrase: Optional[str] = None,
         port: str = "22",
         command_inspectors: Optional[list[CommandInspector]] = None,
-        custom_environment: Optional[dict] = None
+        custom_environment: Optional[dict] = None,
     ) -> None:
         super().__init__()
         self.connection_provider = SshConnectionProvider()
-        self.connection_provider.store_creds(
-            host, SshCredentials(login, password, private_key_path, private_key_passphrase)
-        )
+        self.connection_provider.store_creds(host, SshCredentials(login, password, private_key_path, private_key_passphrase))
         self.host = host
         self.port = port
 
@@ -220,9 +213,7 @@ class SSHShell(Shell):
             result = self._exec_non_interactive(command, options)
 
         if options.check and result.return_code != 0:
-            raise RuntimeError(
-                f"Command: {command}\nreturn code: {result.return_code}\nOutput: {result.stdout}\nStderr: {result.stderr}\n"
-            )
+            raise RuntimeError(f"Command: {command}\nreturn code: {result.return_code}\nOutput: {result.stdout}\nStderr: {result.stderr}\n")
         return result
 
     @log_command

src/frostfs_testlib/steps/http_gate.py

@@ -33,6 +33,7 @@ def get_via_http_gate(
     oid: str,
     node: ClusterNode,
     request_path: Optional[str] = None,
+    presigned_url: Optional[str] = None,
     timeout: Optional[int] = 300,
 ):
     """
@@ -47,6 +48,9 @@ def get_via_http_gate(
     if request_path:
         request = f"{node.http_gate.get_endpoint()}{request_path}"
 
+    if presigned_url:
+        request = presigned_url
+
     response = requests.get(request, stream=True, timeout=timeout, verify=False)
 
     if not response.ok:

src/frostfs_testlib/steps/metrics.py

@@ -6,7 +6,7 @@ from frostfs_testlib.testing.test_control import wait_for_success
 
 
 @reporter.step("Check metrics result")
-@wait_for_success(interval=10)
+@wait_for_success(max_wait_time=300, interval=10)
 def check_metrics_counter(
     cluster_nodes: list[ClusterNode],
     operator: str = "==",
@@ -19,7 +19,7 @@ def check_metrics_counter(
         counter_act += get_metrics_value(cluster_node, parse_from_command, **metrics_greps)
     assert eval(
         f"{counter_act} {operator} {counter_exp}"
-    ), f"Expected: {counter_exp} {operator} Actual: {counter_act} in nodes: {cluster_nodes}"
+    ), f"Actual: {counter_act} {operator} Expected: {counter_exp} in nodes: {cluster_nodes}"
 
 
 @reporter.step("Get metrics value from node: {node}")

src/frostfs_testlib/storage/controllers/cluster_state_controller.py

@@ -247,23 +247,20 @@ class ClusterStateController:
         if service_type == StorageNode:
             self.wait_after_storage_startup()
 
-    # TODO: Deprecated
     @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
-    @reporter.step("Stop all storage services on cluster")
-    def stop_all_storage_services(self, reversed_order: bool = False):
-        nodes = reversed(self.cluster.cluster_nodes) if reversed_order else self.cluster.cluster_nodes
+    @reporter.step("Restart {service_type} service on {node}")
+    def restart_service_of_type(self, node: ClusterNode, service_type: ServiceClass):
+        service = node.service(service_type)
+        service.restart_service()
 
-        for node in nodes:
-            self.stop_service_of_type(node, StorageNode)
-
-    # TODO: Deprecated
     @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
-    @reporter.step("Stop all S3 gates on cluster")
-    def stop_all_s3_gates(self, reversed_order: bool = False):
-        nodes = reversed(self.cluster.cluster_nodes) if reversed_order else self.cluster.cluster_nodes
+    @reporter.step("Restart all {service_type} services")
+    def restart_services_of_type(self, service_type: type[ServiceClass]):
+        services = self.cluster.services(service_type)
+        parallel([service.restart_service for service in services])
 
-        for node in nodes:
-            self.stop_service_of_type(node, S3Gate)
+        if service_type == StorageNode:
+            self.wait_after_storage_startup()
 
     # TODO: Deprecated
     @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
@@ -277,30 +274,6 @@ class ClusterStateController:
     def start_storage_service(self, node: ClusterNode):
         self.start_service_of_type(node, StorageNode)
 
-    # TODO: Deprecated
-    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
-    @reporter.step("Start stopped storage services")
-    def start_stopped_storage_services(self):
-        self.start_stopped_services_of_type(StorageNode)
-
-    # TODO: Deprecated
-    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
-    @reporter.step("Stop s3 gate on {node}")
-    def stop_s3_gate(self, node: ClusterNode, mask: bool = True):
-        self.stop_service_of_type(node, S3Gate, mask)
-
-    # TODO: Deprecated
-    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
-    @reporter.step("Start s3 gate on {node}")
-    def start_s3_gate(self, node: ClusterNode):
-        self.start_service_of_type(node, S3Gate)
-
-    # TODO: Deprecated
-    @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
-    @reporter.step("Start stopped S3 gates")
-    def start_stopped_s3_gates(self):
-        self.start_stopped_services_of_type(S3Gate)
-
     @run_optionally(optionals.OPTIONAL_FAILOVER_ENABLED)
     @reporter.step("Suspend {process_name} service in {node}")
     def suspend_service(self, process_name: str, node: ClusterNode):
@@ -392,19 +365,29 @@ class ClusterStateController:
         shell = node.host.get_shell()
         return datetime.strptime(shell.exec('date +"%Y-%m-%d %H:%M:%S"').stdout.strip(), "%Y-%m-%d %H:%M:%S")
 
-    @reporter.step("Set node time to {in_date}")
+    @reporter.step("Set time on nodes in {in_date}")
+    def change_date_on_all_nodes(self, cluster: Cluster, in_date: datetime) -> None:
+        parallel(self.change_node_date, cluster.cluster_nodes, in_date=in_date)
+
+    @reporter.step("Set time on {node} to {in_date}")
     def change_node_date(self, node: ClusterNode, in_date: datetime) -> None:
         shell = node.host.get_shell()
         in_date_frmt = in_date.strftime("%Y-%m-%d %H:%M:%S")
         shell.exec(f"timedatectl set-time '{in_date_frmt}'")
         node_time = self.get_node_date(node)
+
         with reporter.step(f"Verify difference between {node_time} and {in_date} is less than a minute"):
             assert (node_time - in_date).total_seconds() < 60
 
-    @reporter.step("Restore time")
+    @reporter.step("Restore time on nodes")
+    def restore_date_on_all_nodes(self, cluster: Cluster) -> None:
+        parallel(self.restore_node_date, cluster.cluster_nodes)
+
+    @reporter.step("Restore time on {node}")
     def restore_node_date(self, node: ClusterNode) -> None:
         shell = node.host.get_shell()
         now_time = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+
         with reporter.step(f"Set {now_time} time"):
             shell.exec(f"timedatectl set-time '{now_time}'")
 

src/frostfs_testlib/storage/dataclasses/metrics.py

@@ -1,3 +1,9 @@
+import time
+from functools import wraps
+from typing import Callable
+
+import pytest
+
 from frostfs_testlib.hosting import Host
 from frostfs_testlib.shell.interfaces import CommandResult
 
@@ -7,11 +13,11 @@ class Metrics:
         self.storage = StorageMetrics(host, metrics_endpoint)
 
 
-
 class StorageMetrics:
     """
     Class represents storage metrics in a cluster
     """
+
     def __init__(self, host: Host, metrics_endpoint: str) -> None:
         self.host = host
         self.metrics_endpoint = metrics_endpoint
@@ -29,8 +35,46 @@ class StorageMetrics:
         additional_greps = " |grep ".join([grep_command for grep_command in greps.values()])
         result = shell.exec(f"curl -s {self.metrics_endpoint} | grep {additional_greps}")
         return result
-    
+
     def get_all_metrics(self) -> CommandResult:
         shell = self.host.get_shell()
         result = shell.exec(f"curl -s {self.metrics_endpoint}")
         return result
+
+
+def wait_until_metric_result_is_stable(
+    relative_deviation: float = None, absolute_deviation: int = None, max_attempts: int = 10, sleep_interval: int = 30
+):
+    """
+    A decorator function that repeatedly calls the decorated function until its result stabilizes
+    within a specified relative tolerance or until the maximum number of attempts is reached.
+
+    This decorator is useful for scenarios where a function returns a metric or value that may fluctuate
+    over time, and you want to ensure that the result has stabilized before proceeding.
+    """
+
+    def decorator(func: Callable):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            last_result = None
+            for _ in range(max_attempts):
+                # first function call
+                first_result = func(*args, **kwargs)
+
+                # waiting before the second call
+                time.sleep(sleep_interval)
+
+                # second function call
+                last_result = func(*args, **kwargs)
+
+                # checking value stability
+                if first_result == pytest.approx(last_result, rel=relative_deviation, abs=absolute_deviation):
+                    return last_result
+
+            # if stability is not achieved, return the last value
+            if last_result is not None:
+                return last_result
+
+        return wrapper
+
+    return decorator

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

@@ -4,7 +4,7 @@ from typing import List, Optional
 from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
 from frostfs_testlib.cli.netmap_parser import NetmapParser
 from frostfs_testlib.resources.cli import CLI_DEFAULT_TIMEOUT
-from frostfs_testlib.storage.dataclasses.storage_object_info import NodeNetInfo, NodeNetmapInfo
+from frostfs_testlib.storage.dataclasses.storage_object_info import NodeInfo, NodeNetInfo, NodeNetmapInfo
 
 from .. import interfaces
 
@@ -86,7 +86,7 @@ class NetmapOperations(interfaces.NetmapInterface):
         trace: Optional[bool] = True,
         xhdr: Optional[dict] = None,
         timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
-    ) -> NodeNetmapInfo:
+    ) -> NodeInfo:
         """
         Get target node info.
         """

src/frostfs_testlib/storage/grpc_operations/interfaces/netmap.py

@@ -1,7 +1,7 @@
 from abc import ABC, abstractmethod
 from typing import List, Optional
 
-from frostfs_testlib.storage.dataclasses.storage_object_info import NodeNetInfo, NodeNetmapInfo
+from frostfs_testlib.storage.dataclasses.storage_object_info import NodeInfo, NodeNetInfo, NodeNetmapInfo
 
 
 class NetmapInterface(ABC):
@@ -50,7 +50,7 @@ class NetmapInterface(ABC):
         ttl: Optional[int] = None,
         xhdr: Optional[dict] = None,
         timeout: Optional[str] = None,
-    ) -> NodeNetmapInfo:
+    ) -> NodeInfo:
         """
         Get target node info.
         """

src/frostfs_testlib/utils/cli_utils.py

@@ -68,7 +68,7 @@ def _attach_allure_log(cmd: str, output: str, return_code: int, start_time: date
         f"COMMAND: '{cmd}'\n"
         f"OUTPUT:\n {output}\n"
         f"RC: {return_code}\n"
-        f"Start / End / Elapsed\t {start_time.time()} / {end_time.time()} / {end_time - start_time}"
+        f"Start / End / Elapsed\t {start_time} / {end_time} / {end_time - start_time}"
     )
     with reporter.step(f'COMMAND: {shorten(cmd, width=60, placeholder="...")}'):
         reporter.attach(command_attachment, "Command execution")