[]-Condition-tests

Signed-off-by: Roman Chernykh <r.chernykh@yadro.com>

src/frostfs_testlib/cli/frostfs_cli/shards.py

@@ -241,3 +241,21 @@ class FrostfsCliShards(CliCommand):
             "control shards evacuation status",
             **{param: value for param, value in locals().items() if param not in ["self"]},
         )
+
+    def detach(self, endpoint: str, address: Optional[str] = None, id: Optional[str] = None, timeout: Optional[str] = None):
+        """
+        Detach and close the shards
+
+        Args:
+            address: Address of wallet account
+            endpoint: Remote node control address (as 'multiaddr' or '<host>:<port>')
+            id: List of shard IDs in base58 encoding
+            timeout: Timeout for an operation (default 15s)
+
+        Returns:
+            Command's result.
+        """
+        return self._execute(
+            "control shards detach",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )

src/frostfs_testlib/resources/optionals.py

@@ -16,11 +16,10 @@ OPTIONAL_NODE_UNDER_LOAD = os.getenv("OPTIONAL_NODE_UNDER_LOAD")
 OPTIONAL_FAILOVER_ENABLED = str_to_bool(os.getenv("OPTIONAL_FAILOVER_ENABLED", "true"))
 
 # Set this to True to disable background load. I.E. node which supposed to be stopped will not be actually stopped.
-OPTIONAL_BACKGROUND_LOAD_ENABLED = str_to_bool(
-    os.getenv("OPTIONAL_BACKGROUND_LOAD_ENABLED", "true")
-)
+OPTIONAL_BACKGROUND_LOAD_ENABLED = str_to_bool(os.getenv("OPTIONAL_BACKGROUND_LOAD_ENABLED", "true"))
 
 # Set this to False for disable autouse fixture like node healthcheck during developing time.
-OPTIONAL_AUTOUSE_FIXTURES_ENABLED = str_to_bool(
-    os.getenv("OPTIONAL_AUTOUSE_FIXTURES_ENABLED", "true")
-)
+OPTIONAL_AUTOUSE_FIXTURES_ENABLED = str_to_bool(os.getenv("OPTIONAL_AUTOUSE_FIXTURES_ENABLED", "true"))
+
+# Use cache for fixtures with @cachec_fixture decorator
+OPTIONAL_CACHE_FIXTURES = str_to_bool(os.getenv("OPTIONAL_CACHE_FIXTURES", "false"))

src/frostfs_testlib/s3/aws_cli_client.py

@@ -196,11 +196,20 @@ class AwsCliClient(S3ClientWrapper):
         return response.get("LocationConstraint")
 
     @reporter.step("List objects S3")
-    def list_objects(self, bucket: str, full_output: bool = False) -> Union[dict, list[str]]:
+    def list_objects(
+        self, bucket: str, page_size: Optional[int] = None, prefix: Optional[str] = None, full_output: bool = False
+    ) -> Union[dict, list[str]]:
         if bucket.startswith("-") or " " in bucket:
             bucket = f'"{bucket}"'
 
-        cmd = f"aws {self.common_flags} s3api list-objects --bucket {bucket} " f"--endpoint {self.s3gate_endpoint} --profile {self.profile}"
+        cmd = f"aws {self.common_flags} s3api list-objects --bucket {bucket} " f"--endpoint {self.s3gate_endpoint} "
+        if page_size:
+            cmd = cmd.replace("--no-paginate", "")
+            cmd += f" --page-size {page_size} "
+        if prefix:
+            cmd += f" --prefix {prefix}"
+        if self.profile:
+            cmd += f" --profile {self.profile} "
         output = self.local_shell.exec(cmd).stdout
         response = self._to_json(output)
 
@@ -1440,3 +1449,90 @@ class AwsCliClient(S3ClientWrapper):
         response = self._to_json(output)
 
         return response
+
+    # MFA METHODS
+    @reporter.step("Creates a new virtual MFA device")
+    def iam_create_virtual_mfa_device(self, virtual_mfa_device_name: str, outfile: str, bootstrap_method: str) -> tuple:
+        cmd = f"aws {self.common_flags} iam create-virtual-mfa-device --virtual-mfa-device-name {virtual_mfa_device_name}\
+              --outfile {outfile} --bootstrap-method {bootstrap_method} --endpoint {self.iam_endpoint}"
+
+        if self.profile:
+            cmd += f" --profile {self.profile}"
+
+        output = self.local_shell.exec(cmd).stdout
+        response = self._to_json(output)
+        serial_number = response.get("VirtualMFADevice", {}).get("SerialNumber")
+        assert serial_number, f"Expected SerialNumber in response:\n{response}"
+
+        return serial_number, False
+
+    @reporter.step("Deactivates the specified MFA device and removes it from association with the user name")
+    def iam_deactivate_mfa_device(self, user_name: str, serial_number: str) -> dict:
+        cmd = f"aws {self.common_flags} iam deactivate-mfa-device --user-name {user_name} --serial-number {serial_number} --endpoint {self.iam_endpoint}"
+        if self.profile:
+            cmd += f" --profile {self.profile}"
+
+        output = self.local_shell.exec(cmd).stdout
+        response = self._to_json(output)
+
+        return response
+
+    @reporter.step("Deletes a virtual MFA device")
+    def iam_delete_virtual_mfa_device(self, serial_number: str) -> dict:
+        cmd = f"aws {self.common_flags} iam delete-virtual-mfa-device --serial-number {serial_number} --endpoint {self.iam_endpoint}"
+        if self.profile:
+            cmd += f" --profile {self.profile}"
+
+        output = self.local_shell.exec(cmd).stdout
+        response = self._to_json(output)
+
+        return response
+
+    @reporter.step("Enables the specified MFA device and associates it with the specified IAM user")
+    def iam_enable_mfa_device(self, user_name: str, serial_number: str, authentication_code1: str, authentication_code2: str) -> dict:
+        cmd = f"aws {self.common_flags} iam enable-mfa-device --user-name {user_name} --serial-number {serial_number} --authentication-code1 {authentication_code1}\
+              --authentication-code2 {authentication_code2} --endpoint {self.iam_endpoint}"
+        if self.profile:
+            cmd += f" --profile {self.profile}"
+
+        output = self.local_shell.exec(cmd).stdout
+        response = self._to_json(output)
+
+        return response
+
+    @reporter.step("Lists the MFA devices for an IAM user")
+    def iam_list_virtual_mfa_devices(self) -> dict:
+        cmd = f"aws {self.common_flags} iam list-virtual-mfa-devices --endpoint {self.iam_endpoint}"
+        if self.profile:
+            cmd += f" --profile {self.profile}"
+
+        output = self.local_shell.exec(cmd).stdout
+        response = self._to_json(output)
+        assert response.get("VirtualMFADevices"), f"Expected VirtualMFADevices in response:\n{response}"
+
+        return response
+
+    @reporter.step("Get session token for user")
+    def sts_get_session_token(
+        self, duration_seconds: Optional[str] = None, serial_number: Optional[str] = None, token_code: Optional[str] = None
+    ) -> tuple:
+        cmd = f"aws {self.common_flags} sts get-session-token   --endpoint {self.iam_endpoint}"
+        if duration_seconds:
+            cmd += f" --duration-seconds {duration_seconds}"
+        if serial_number:
+            cmd += f" --serial-number {serial_number}"
+        if token_code:
+            cmd += f" --token-code {token_code}"
+        if self.profile:
+            cmd += f" --profile {self.profile}"
+
+        output = self.local_shell.exec(cmd).stdout
+        response = self._to_json(output)
+        access_key = response.get("Credentials", {}).get("AccessKeyId")
+        secret_access_key = response.get("Credentials", {}).get("SecretAccessKey")
+        session_token = response.get("Credentials", {}).get("SessionToken")
+        assert access_key, f"Expected AccessKeyId in response:\n{response}"
+        assert secret_access_key, f"Expected SecretAccessKey in response:\n{response}"
+        assert session_token, f"Expected SessionToken in response:\n{response}"
+
+        return access_key, secret_access_key, session_token

src/frostfs_testlib/s3/boto3_client.py

@@ -41,6 +41,8 @@ class Boto3ClientWrapper(S3ClientWrapper):
         self.boto3_iam_client: S3Client = None
         self.iam_endpoint: str = ""
 
+        self.boto3_sts_client: S3Client = None
+
         self.access_key_id: str = access_key_id
         self.secret_access_key: str = secret_access_key
         self.profile = profile
@@ -87,6 +89,14 @@ class Boto3ClientWrapper(S3ClientWrapper):
             endpoint_url=self.iam_endpoint,
             verify=False,
         )
+        # since the STS does not have an enpoint, IAM is used
+        self.boto3_sts_client = self.session.client(
+            service_name="sts",
+            aws_access_key_id=self.access_key_id,
+            aws_secret_access_key=self.secret_access_key,
+            endpoint_url=iam_endpoint,
+            verify=False,
+        )
 
     def _to_s3_param(self, param: str) -> str:
         replacement_map = {
@@ -388,10 +398,17 @@ class Boto3ClientWrapper(S3ClientWrapper):
         return response if full_output else obj_list
 
     @reporter.step("List objects S3")
-    def list_objects(self, bucket: str, full_output: bool = False) -> Union[dict, list[str]]:
+    def list_objects(
+        self, bucket: str, page_size: Optional[int] = None, prefix: Optional[str] = None, full_output: bool = False
+    ) -> Union[dict, list[str]]:
+        params = {"Bucket": bucket}
+        if page_size:
+            params["MaxKeys"] = page_size
+        if prefix:
+            params["Prefix"] = prefix
         response = self._exec_request(
             self.boto3_client.list_objects,
-            params={"Bucket": bucket},
+            params,
             endpoint=self.s3gate_endpoint,
             profile=self.profile,
         )
@@ -1265,3 +1282,67 @@ class Boto3ClientWrapper(S3ClientWrapper):
             endpoint=self.iam_endpoint,
             profile=self.profile,
         )
+
+        # MFA methods
+
+    @reporter.step("Creates a new virtual MFA device")
+    def iam_create_virtual_mfa_device(
+        self, virtual_mfa_device_name: str, outfile: Optional[str] = None, bootstrap_method: Optional[str] = None
+    ) -> tuple:
+        response = self.boto3_iam_client.create_virtual_mfa_device(VirtualMFADeviceName=virtual_mfa_device_name)
+
+        serial_number = response.get("VirtualMFADevice", {}).get("SerialNumber")
+        base32StringSeed = response.get("VirtualMFADevice", {}).get("Base32StringSeed")
+        assert serial_number, f"Expected SerialNumber in response:\n{response}"
+        assert base32StringSeed, f"Expected Base32StringSeed in response:\n{response}"
+
+        return serial_number, base32StringSeed
+
+    @reporter.step("Deactivates the specified MFA device and removes it from association with the user name")
+    def iam_deactivate_mfa_device(self, user_name: str, serial_number: str) -> dict:
+        response = self.boto3_iam_client.deactivate_mfa_device(UserName=user_name, SerialNumber=serial_number)
+
+        return response
+
+    @reporter.step("Deletes a virtual MFA device")
+    def iam_delete_virtual_mfa_device(self, serial_number: str) -> dict:
+        response = self.boto3_iam_client.delete_virtual_mfa_device(SerialNumber=serial_number)
+
+        return response
+
+    @reporter.step("Enables the specified MFA device and associates it with the specified IAM user")
+    def iam_enable_mfa_device(self, user_name: str, serial_number: str, authentication_code1: str, authentication_code2: str) -> dict:
+        response = self.boto3_iam_client.enable_mfa_device(
+            UserName=user_name,
+            SerialNumber=serial_number,
+            AuthenticationCode1=authentication_code1,
+            AuthenticationCode2=authentication_code2,
+        )
+
+        return response
+
+    @reporter.step("Lists the MFA devices for an IAM user")
+    def iam_list_virtual_mfa_devices(self) -> dict:
+        response = self.boto3_iam_client.list_virtual_mfa_devices()
+        assert response.get("VirtualMFADevices"), f"Expected VirtualMFADevices in response:\n{response}"
+
+        return response
+
+    @reporter.step("Get session token for user")
+    def sts_get_session_token(
+        self, duration_seconds: Optional[str] = "", serial_number: Optional[str] = "", token_code: Optional[str] = ""
+    ) -> tuple:
+        response = self.boto3_sts_client.get_session_token(
+            DurationSeconds=duration_seconds,
+            SerialNumber=serial_number,
+            TokenCode=token_code,
+        )
+
+        access_key = response.get("Credentials", {}).get("AccessKeyId")
+        secret_access_key = response.get("Credentials", {}).get("SecretAccessKey")
+        session_token = response.get("Credentials", {}).get("SessionToken")
+        assert access_key, f"Expected AccessKeyId in response:\n{response}"
+        assert secret_access_key, f"Expected SecretAccessKey in response:\n{response}"
+        assert session_token, f"Expected SessionToken in response:\n{response}"
+
+        return access_key, secret_access_key, session_token

src/frostfs_testlib/s3/interfaces.py

@@ -195,7 +195,9 @@ class S3ClientWrapper(HumanReadableABC):
         """
 
     @abstractmethod
-    def list_objects(self, bucket: str, full_output: bool = False) -> Union[dict, list[str]]:
+    def list_objects(
+        self, bucket: str, page_size: Optional[int] = None, prefix: Optional[str] = None, full_output: bool = False
+    ) -> Union[dict, list[str]]:
         """Returns some or all (up to 1,000) of the objects in a bucket with each request.
         You can use the request parameters as selection criteria to return a subset of the objects in a bucket.
         A 200 OK response can contain valid or invalid XML. Make sure to design your application
@@ -578,3 +580,32 @@ class S3ClientWrapper(HumanReadableABC):
     @abstractmethod
     def iam_untag_user(self, user_name: str, tag_keys: list) -> dict:
         """Removes the specified tags from the user"""
+
+    # MFA methods
+    @abstractmethod
+    def iam_create_virtual_mfa_device(
+        self, virtual_mfa_device_name: str, outfile: Optional[str] = None, bootstrap_method: Optional[str] = None
+    ) -> tuple:
+        """Creates a new virtual MFA device"""
+
+    @abstractmethod
+    def iam_deactivate_mfa_device(self, user_name: str, serial_number: str) -> dict:
+        """Deactivates the specified MFA device and removes it from association with the user name"""
+
+    @abstractmethod
+    def iam_delete_virtual_mfa_device(self, serial_number: str) -> dict:
+        """Deletes a virtual MFA device"""
+
+    @abstractmethod
+    def iam_enable_mfa_device(self, user_name: str, serial_number: str, authentication_code1: str, authentication_code2: str) -> dict:
+        """Enables the specified MFA device and associates it with the specified IAM user"""
+
+    @abstractmethod
+    def iam_list_virtual_mfa_devices(self) -> dict:
+        """Lists the MFA devices for an IAM user"""
+
+    @abstractmethod
+    def sts_get_session_token(
+        self, duration_seconds: Optional[str] = None, serial_number: Optional[str] = None, token_code: Optional[str] = None
+    ) -> tuple:
+        """Get session token for user"""

src/frostfs_testlib/testing/test_control.py

@@ -1,13 +1,16 @@
 import inspect
 import logging
+import os
 from functools import wraps
 from time import sleep, time
 from typing import Any
 
+import yaml
 from _pytest.outcomes import Failed
 from pytest import fail
 
 from frostfs_testlib import reporter
+from frostfs_testlib.resources.common import ASSETS_DIR
 from frostfs_testlib.utils.func_utils import format_by_args
 
 logger = logging.getLogger("NeoLogger")
@@ -128,6 +131,42 @@ def run_optionally(enabled: bool, mock_value: Any = True):
     return deco
 
 
+def cached_fixture(enabled: bool):
+    """
+    Decorator to cache fixtures.
+    MUST be placed after @pytest.fixture and before @allure decorators.
+
+    Args:
+    enabled: if true, decorated func will be cached.
+    """
+
+    def deco(func):
+        @wraps(func)
+        def func_impl(*a, **kw):
+            # TODO: *a and *kw should be parsed to some kind of hashsum and used in filename to prevent cache load from different parameters
+            cache_file = os.path.join(ASSETS_DIR, f"fixture_cache_{func.__name__}.yml")
+
+            if enabled and os.path.exists(cache_file):
+                with open(cache_file, "r") as cache_input:
+                    return yaml.load(cache_input, Loader=yaml.Loader)
+
+            result = func(*a, **kw)
+
+            if enabled:
+                with open(cache_file, "w") as cache_output:
+                    yaml.dump(result, cache_output)
+            return result
+
+        # TODO: cache yielding fixtures
+        @wraps(func)
+        def gen_impl(*a, **kw):
+            raise NotImplementedError("Not implemented for yielding fixtures")
+
+        return gen_impl if inspect.isgeneratorfunction(func) else func_impl
+
+    return deco
+
+
 def wait_for_success(
     max_wait_time: int = 60,
     interval: int = 1,

src/frostfs_testlib/utils/file_utils.py

@@ -45,7 +45,7 @@ def ensure_directory_opener(path, flags):
 # TODO: Do not add {size} to title yet, since it produces dynamic info in top level steps
 # Use object_size dt in future as argument
 @reporter.step("Generate file")
-def generate_file(size: int) -> TestFile:
+def generate_file(size: int, file_name: Optional[str] = None) -> TestFile:
     """Generates a binary file with the specified size in bytes.
 
     Args:
@@ -54,7 +54,11 @@ def generate_file(size: int) -> TestFile:
     Returns:
         The path to the generated file.
     """
-    test_file = TestFile(os.path.join(ASSETS_DIR, string_utils.unique_name("object-")))
+
+    if file_name is None:
+        file_name = string_utils.unique_name("object-")
+
+    test_file = TestFile(os.path.join(ASSETS_DIR, file_name))
     with open(test_file, "wb", opener=ensure_directory_opener) as file:
         file.write(os.urandom(size))
     logger.info(f"File with size {size} bytes has been generated: {test_file}")