tests: add positive tests for s3 presigned urls

neofs-testlib/neofs_testlib/cli/neofs_authmate/authmate.py

@@ -2,13 +2,16 @@
 
 from neofs_testlib.cli.neofs_authmate.secret import NeofsAuthmateSecret
 from neofs_testlib.cli.neofs_authmate.version import NeofsAuthmateVersion
+from neofs_testlib.cli.neofs_authmate.presigned import NeofsAuthmatePresigned
 from neofs_testlib.shell import Shell
 
 
 class NeofsAuthmate:
     secret: Optional[NeofsAuthmateSecret] = None
     version: Optional[NeofsAuthmateVersion] = None
+    presigned: Optional[NeofsAuthmatePresigned] = None
 
     def __init__(self, shell: Shell, neofs_authmate_exec_path: str):
         self.secret = NeofsAuthmateSecret(shell, neofs_authmate_exec_path)
         self.version = NeofsAuthmateVersion(shell, neofs_authmate_exec_path)
+        self.presigned = NeofsAuthmatePresigned(shell, neofs_authmate_exec_path)

neofs-testlib/neofs_testlib/cli/neofs_authmate/presigned.py

@@ -0,0 +1,33 @@
+from neofs_testlib.cli.cli_command import CliCommand
+from neofs_testlib.shell import CommandResult
+
+
+class NeofsAuthmatePresigned(CliCommand):
+    def generate_presigned_url(
+        self,
+        endpoint: str,
+        method: str,
+        bucket: str,
+        object: str,
+        lifetime: str,
+        aws_secret_access_key: str,
+        aws_access_key_id: str,
+    ) -> CommandResult:
+        """Generate presigned URL
+
+        Args:
+            endpoint: Endpoint of s3-gw
+            method: HTTP method to perform action
+            bucket: Bucket name to perform action
+            object: Object name to perform action
+            lifetime: Lifetime of presigned URL
+            aws-access-key-id: AWS access key id to sign the URL
+            aws-secret-access-key: AWS access secret access key to sign the URL
+
+        Returns:
+            Command's result.
+        """
+        return self._execute(
+            "generate-presigned-url",
+            **{param: param_value for param, param_value in locals().items() if param not in ["self"]},
+        )

pytest_tests/lib/helpers/aws_cli_client.py

@@ -543,6 +543,13 @@ def put_bucket_ownership_controls(self, Bucket, OwnershipControls):
         output = _cmd_run(cmd)
         return self._to_json(output)
 
+    def generate_presigned_url(self, ClientMethod: str, Params: dict, ExpiresIn: int, HttpMethod: str):
+        cmd = (
+            f"aws s3 {self.common_flags} --endpoint {self.s3gate_endpoint} presign "
+            f"s3://{Params["Bucket"]}/{Params["Key"]} --expires-in {ExpiresIn}"
+        )
+        return _cmd_run(cmd)
+
     @staticmethod
     def _to_json(output: str) -> dict:
         json_output = {}

pytest_tests/lib/s3/s3_base.py

@@ -185,7 +185,8 @@ def configure_boto3_client(access_key_id: str, secret_access_key: str, s3gate_en
             retries={
                 "max_attempts": MAX_REQUEST_ATTEMPTS,
                 "mode": RETRY_MODE,
-            }
+            },
+            signature_version="v4",
         )
 
         s3_client = session.client(

pytest_tests/tests/s3/test_s3_presigned.py

@@ -0,0 +1,76 @@
+import json
+import logging
+import os
+import shutil
+import uuid
+
+import allure
+import pytest
+import requests
+from helpers.common import TEST_FILES_DIR, get_assets_dir_path
+from helpers.file_helper import generate_file, get_file_hash
+from helpers.s3_helper import object_key_from_file_path
+from neofs_testlib.cli import NeofsAuthmate
+from s3 import s3_object
+from s3.s3_base import TestNeofsS3Base
+
+logger = logging.getLogger("NeoLogger")
+
+
+def pytest_generate_tests(metafunc):
+    if "s3_client" in metafunc.fixturenames:
+        metafunc.parametrize("s3_client", ["aws cli", "boto3"], indirect=True)
+
+
+class TestS3Presigned(TestNeofsS3Base):
+    @allure.title("Test S3: Get Object With Presigned Url")
+    @pytest.mark.parametrize("url_from", ["neofs_authmate", "s3"])
+    def test_s3_get_object_with_presigned_url(self, bucket, simple_object_size, url_from: str):
+        file_path = generate_file(simple_object_size)
+        file_name = object_key_from_file_path(file_path)
+
+        with allure.step("Put object into Bucket"):
+            s3_object.put_object_s3(self.s3_client, bucket, file_path)
+
+        with allure.step(f"Get presigned URL from {url_from}"):
+            if url_from == "neofs_authmate":
+                neofs_authmate = NeofsAuthmate(
+                    shell=self.shell, neofs_authmate_exec_path=self.neofs_env.neofs_s3_authmate_path
+                )
+                raw_url = json.loads(
+                    neofs_authmate.presigned.generate_presigned_url(
+                        endpoint=f"https://{self.neofs_env.s3_gw.address}",
+                        method="GET",
+                        bucket=bucket,
+                        object=file_name,
+                        lifetime="30s",
+                        aws_secret_access_key=self.secret_access_key,
+                        aws_access_key_id=self.access_key_id,
+                    ).stdout
+                )
+                presigned_url = raw_url["URL"]
+            else:
+                presigned_url = self.s3_client.generate_presigned_url(
+                    ClientMethod="get_object",
+                    Params={"Bucket": bucket, "Key": file_name},
+                    ExpiresIn=30,
+                    HttpMethod="GET",
+                ).strip()
+                logger.info(f"Presigned URL: {presigned_url}")
+
+        with allure.step("Get object with generated presigned url"):
+            resp = requests.get(presigned_url, stream=True, timeout=30, verify=False)
+
+            if not resp.ok:
+                raise Exception(
+                    f"""Failed to get object via presigned url:
+                        request: {resp.request.path_url},
+                        response: {resp.text},
+                        status code: {resp.status_code} {resp.reason}"""
+                )
+
+            new_file_path = os.path.join(get_assets_dir_path(), TEST_FILES_DIR, f"temp_file_{uuid.uuid4()}")
+            with open(new_file_path, "wb") as file:
+                shutil.copyfileobj(resp.raw, file)
+
+            assert get_file_hash(file_path) == get_file_hash(new_file_path), "Files hashes are different"