[#353] Extend testsuites for PATCH method

Expandable suites:
- TestApeContainer
- TestApeBearer
- TestApeLocalOverrideAllow
- TestApeLocalOverrideDeny
- TestObjectApiWithoutUser
- TestObjectApiWithBearerToken

Signed-off-by: Kirill Sosnovskikh <k.sosnovskikh@yadro.com>

CODEOWNERS

@@ -1 +1,3 @@
-*       @JuliaKovshova @abereziny @d.zayakin @anikeev-yadro @anurindm @ylukoyan @i.niyazov
+.*          @TrueCloudLab/qa-committers
+.forgejo/.* @potyarkin
+Makefile    @potyarkin

pytest_tests/helpers/container_access.py

@@ -20,7 +20,10 @@ ALL_OBJECT_OPERATIONS = ape.ObjectOperations.get_all()
 
 FULL_ACCESS = {op: True for op in ALL_OBJECT_OPERATIONS}
 NO_ACCESS = {op: False for op in ALL_OBJECT_OPERATIONS}
-RO_ACCESS = {op: True if op not in [ape.ObjectOperations.PUT, ape.ObjectOperations.DELETE] else False for op in ALL_OBJECT_OPERATIONS}
+RO_ACCESS = {
+    op: True if op not in [ape.ObjectOperations.PUT, ape.ObjectOperations.DELETE, ape.ObjectOperations.PATCH] else False
+    for op in ALL_OBJECT_OPERATIONS
+}
 
 
 def assert_access_to_container(

pytest_tests/helpers/container_creation.py

@@ -63,7 +63,15 @@ def _create_container_by_spec(
     cluster: Cluster,
     endpoint: str,
 ) -> str:
-    return create_container(wallet, shell, endpoint, container_request.parsed_rule(cluster), wait_for_creation=False)
+    return create_container(
+        wallet,
+        shell,
+        endpoint,
+        container_request.parsed_rule(cluster),
+        wait_for_creation=False,
+        nns_zone=container_request.ns_zone,
+        nns_name=container_request.ns_name,
+    )
 
 
 def _apply_ape_rules(cid: str, frostfs_cli: FrostfsCli, endpoint: str, ape_rules: list[ape.Rule]):

pytest_tests/helpers/container_request.py

@@ -21,6 +21,9 @@ class ContainerRequest:
 
     short_name: str | None = None
 
+    ns_name: str | None = None
+    ns_zone: str | None = None
+
     def __post_init__(self):
         if self.ape_rules is None:
             self.ape_rules = []

pytest_tests/testsuites/access/ape/test_ape.py

@@ -2,10 +2,12 @@ import allure
 import pytest
 from frostfs_testlib import reporter
 from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
+from frostfs_testlib.resources.error_patterns import OBJECT_ACCESS_DENIED
 from frostfs_testlib.steps.cli.object import put_object_to_random_node
 from frostfs_testlib.steps.node_management import drop_object
 from frostfs_testlib.storage.dataclasses import ape
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
+from frostfs_testlib.storage.grpc_operations.interfaces import GrpcClientWrapper
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 from frostfs_testlib.utils import wallet_utils
 from frostfs_testlib.utils.failover_utils import wait_object_replication
@@ -33,9 +35,12 @@ def allowed_wallet(default_wallet: WalletInfo, other_wallet: WalletInfo, role: a
 @pytest.mark.nightly
 @pytest.mark.ape
 class TestApeContainer(ClusterTestBase):
+    # TODO: Without PATCH operation,
+    # since it requires specific permissions that do not apply when testing all operations at once
     @pytest.mark.sanity
     @allure.title("Deny operations via APE by role (role={role}, obj_size={object_size})")
     @pytest.mark.parametrize("role", [ape.Role.OWNER, ape.Role.OTHERS], indirect=True)
+    @pytest.mark.parametrize("objects", [4], indirect=True)
     def test_deny_operations_via_ape_by_role(
         self,
         denied_wallet: WalletInfo,
@@ -44,7 +49,7 @@ class TestApeContainer(ClusterTestBase):
         container: str,
         objects: list[str],
         role: ape.Role,
-        file_path: TestFile,
+        test_file: TestFile,
         rpc_endpoint: str,
     ):
         with reporter.step(f"Deny all operations for {role} via APE"):
@@ -58,10 +63,10 @@ class TestApeContainer(ClusterTestBase):
 
         with reporter.step(f"Assert denied role have no access to public container"):
             # access checks will try to remove object, so we use .pop() to ensure we have object before deletion
-            assert_no_access_to_container(denied_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
+            assert_no_access_to_container(denied_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
         with reporter.step(f"Assert allowed role have full access to public container"):
-            assert_full_access_to_container(allowed_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(allowed_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
         with reporter.step(f"Remove deny rule from APE"):
             frostfs_cli.ape_manager.remove(rpc_endpoint, deny_rule.chain_id, target_name=container, target_type="container")
@@ -70,12 +75,15 @@ class TestApeContainer(ClusterTestBase):
             self.wait_for_blocks()
 
         with reporter.step("Assert allowed role have full access to public container"):
-            assert_full_access_to_container(allowed_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(allowed_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
         with reporter.step("Assert denied role have full access to public container"):
-            assert_full_access_to_container(denied_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(denied_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
+    # TODO: Without PATCH operation,
+    # since it requires specific permissions that do not apply when testing all operations at once
     @allure.title("Deny operations for others via APE excluding single pubkey (obj_size={object_size})")
+    @pytest.mark.parametrize("objects", [2], indirect=True)
     def test_deny_opeartions_excluding_pubkey(
         self,
         frostfs_cli: FrostfsCli,
@@ -85,7 +93,7 @@ class TestApeContainer(ClusterTestBase):
         container: str,
         objects: list[str],
         rpc_endpoint: str,
-        file_path: TestFile,
+        test_file: TestFile,
     ):
         with reporter.step("Add deny APE rules for others except single wallet"):
             rule_conditions = [
@@ -103,13 +111,13 @@ class TestApeContainer(ClusterTestBase):
 
         with reporter.step("Assert others have no access to public container"):
             # access checks will try to remove object, so we use .pop() to ensure we have object before deletion
-            assert_no_access_to_container(other_wallet, container, objects[0], file_path, self.shell, self.cluster)
+            assert_no_access_to_container(other_wallet, container, objects[0], test_file, self.shell, self.cluster)
 
         with reporter.step("Assert owner have full access to public container"):
-            assert_full_access_to_container(default_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(default_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
         with reporter.step("Assert allowed wallet have full access to public container"):
-            assert_full_access_to_container(other_wallet_2, container, objects.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(other_wallet_2, container, objects.pop(), test_file, self.shell, self.cluster)
 
     @allure.title("Replication works with APE deny rules on OWNER and OTHERS (obj_size={object_size})")
     @pytest.mark.parametrize(
@@ -123,10 +131,10 @@ class TestApeContainer(ClusterTestBase):
         frostfs_cli: FrostfsCli,
         container: str,
         rpc_endpoint: str,
-        file_path: TestFile,
+        test_file: TestFile,
     ):
         with reporter.step("Put object to container"):
-            oid = put_object_to_random_node(default_wallet, file_path, container, self.shell, self.cluster)
+            oid = put_object_to_random_node(default_wallet, test_file, container, self.shell, self.cluster)
 
         with reporter.step("Wait for object replication after upload"):
             wait_object_replication(container, oid, len(self.cluster.cluster_nodes), self.shell, self.cluster.storage_nodes)
@@ -151,10 +159,13 @@ class TestApeContainer(ClusterTestBase):
         with reporter.step("Wait for dropped object to be replicated"):
             wait_object_replication(container, oid, len(self.cluster.storage_nodes), self.shell, self.cluster.storage_nodes)
 
+    # TODO: Without PATCH operation,
+    # since it requires specific permissions that do not apply when testing all operations at once
     @allure.title("Deny operations via APE by role (role=ir, obj_size={object_size})")
     @pytest.mark.parametrize("container_request", [OWNER_ALLOW_ALL], indirect=True)
+    @pytest.mark.parametrize("objects", [3], indirect=True)
     def test_deny_operations_via_ape_by_role_ir(
-        self, frostfs_cli: FrostfsCli, ir_wallet: WalletInfo, container: str, objects: list[str], rpc_endpoint: str, file_path: TestFile
+        self, frostfs_cli: FrostfsCli, ir_wallet: WalletInfo, container: str, objects: list[str], rpc_endpoint: str, test_file: TestFile
     ):
         default_ir_access = {
             ape.ObjectOperations.PUT: False,
@@ -163,11 +174,12 @@ class TestApeContainer(ClusterTestBase):
             ape.ObjectOperations.GET_RANGE: True,
             ape.ObjectOperations.GET_RANGE_HASH: True,
             ape.ObjectOperations.SEARCH: True,
+            ape.ObjectOperations.PATCH: False,
             ape.ObjectOperations.DELETE: False,
         }
 
         with reporter.step("Assert IR wallet access in default state"):
-            assert_access_to_container(default_ir_access, ir_wallet, container, objects[0], file_path, self.shell, self.cluster)
+            assert_access_to_container(default_ir_access, ir_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
         with reporter.step("Add deny APE rule with deny all operations for IR role"):
             rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, [ape.Condition.by_role(ape.Role.IR.value)])
@@ -177,7 +189,7 @@ class TestApeContainer(ClusterTestBase):
             self.wait_for_blocks()
 
         with reporter.step("Assert IR wallet ignores APE rules"):
-            assert_access_to_container(default_ir_access, ir_wallet, container, objects[0], file_path, self.shell, self.cluster)
+            assert_access_to_container(default_ir_access, ir_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
         with reporter.step("Remove APE rule"):
             frostfs_cli.ape_manager.remove(rpc_endpoint, rule.chain_id, target_name=container, target_type="container")
@@ -186,10 +198,13 @@ class TestApeContainer(ClusterTestBase):
             self.wait_for_blocks()
 
         with reporter.step("Assert IR wallet access is restored"):
-            assert_access_to_container(default_ir_access, ir_wallet, container, objects[0], file_path, self.shell, self.cluster)
+            assert_access_to_container(default_ir_access, ir_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
+    # TODO: Without PATCH operation,
+    # since it requires specific permissions that do not apply when testing all operations at once
     @allure.title("Deny operations via APE by role (role=container, obj_size={object_size})")
     @pytest.mark.parametrize("container_request", [OWNER_ALLOW_ALL], indirect=True)
+    @pytest.mark.parametrize("objects", [3], indirect=True)
     def test_deny_operations_via_ape_by_role_container(
         self,
         frostfs_cli: FrostfsCli,
@@ -197,7 +212,7 @@ class TestApeContainer(ClusterTestBase):
         container: str,
         objects: list[str],
         rpc_endpoint: str,
-        file_path: TestFile,
+        test_file: TestFile,
     ):
         access_matrix = {
             ape.ObjectOperations.PUT: True,
@@ -206,11 +221,12 @@ class TestApeContainer(ClusterTestBase):
             ape.ObjectOperations.GET_RANGE: True,
             ape.ObjectOperations.GET_RANGE_HASH: True,
             ape.ObjectOperations.SEARCH: True,
+            ape.ObjectOperations.PATCH: True,
             ape.ObjectOperations.DELETE: True,
         }
 
         with reporter.step("Assert CONTAINER wallet access in default state"):
-            assert_access_to_container(access_matrix, container_node_wallet, container, objects[0], file_path, self.shell, self.cluster)
+            assert_access_to_container(access_matrix, container_node_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
         rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, ape.Condition.by_role(ape.Role.CONTAINER.value))
 
@@ -221,7 +237,7 @@ class TestApeContainer(ClusterTestBase):
             self.wait_for_blocks()
 
         with reporter.step("Assert CONTAINER wallet ignores APE rule"):
-            assert_access_to_container(access_matrix, container_node_wallet, container, objects[1], file_path, self.shell, self.cluster)
+            assert_access_to_container(access_matrix, container_node_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
         with reporter.step("Remove APE rule"):
             frostfs_cli.ape_manager.remove(rpc_endpoint, rule.chain_id, target_name=container, target_type="container")
@@ -230,4 +246,152 @@ class TestApeContainer(ClusterTestBase):
             self.wait_for_blocks()
 
         with reporter.step("Assert CONTAINER wallet access after rule was removed"):
-            assert_access_to_container(access_matrix, container_node_wallet, container, objects[2], file_path, self.shell, self.cluster)
+            assert_access_to_container(access_matrix, container_node_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
+
+    # ^
+    @allure.title("Deny PATCH operation via APE (object_size={object_size})")
+    @pytest.mark.parametrize("objects", [1], indirect=True)
+    def test_patch_object_with_deny_rule(
+        self,
+        frostfs_cli: FrostfsCli,
+        grpc_client: GrpcClientWrapper,
+        grpc_client_with_other_wallet: GrpcClientWrapper,
+        grpc_client_with_container_wallet: GrpcClientWrapper,
+        grpc_client_with_ir_wallet: GrpcClientWrapper,
+        container: str,
+        objects: list[str],
+        test_file: TestFile,
+    ):
+        patch_params = {
+            "cid": container,
+            "oid": objects[0],
+            "endpoint": self.cluster.default_rpc_endpoint,
+            "ranges": ["300:200"],
+            "payloads": [test_file],
+            "new_attrs": "owner=true",
+            "timeout": "200s",
+        }
+
+        with reporter.step("Check that PATCH is available with owner wallet"):
+            patched_oid = grpc_client.object.patch(**patch_params)
+            assert patched_oid != patch_params["oid"], "OID of patched object must be different from original one"
+            patch_params["oid"] = patched_oid
+
+        with reporter.step("Check that PATCH is available with another wallet"):
+            patch_params["ranges"] = ["100:50"]
+            patch_params["new_attrs"] = "other=true"
+
+            patched_oid = grpc_client_with_other_wallet.object.patch(**patch_params)
+            assert patched_oid != patch_params["oid"], "OID of patched object must be different from original one"
+            patch_params["oid"] = patched_oid
+
+        with reporter.step("Check that PATCH is available with container wallet"):
+            patch_params["ranges"] = ["600:0"]
+            patch_params["new_attrs"] = "container=true"
+
+            patched_oid = grpc_client_with_container_wallet.object.patch(**patch_params)
+            assert patched_oid != patch_params["oid"], "OID of patched object must be different from original one"
+            patch_params["oid"] = patched_oid
+
+        with reporter.step("Check that PATCH is available with ir wallet"):
+            patch_params["ranges"] = ["0:1000"]
+            patch_params["new_attrs"] = "ir=true"
+
+            patched_oid = grpc_client_with_ir_wallet.object.patch(**patch_params)
+            assert patched_oid != patch_params["oid"], "OID of patched object must be different from original one"
+            patch_params["oid"] = patched_oid
+
+        rule = ape.Rule(ape.Verb.DENY, ape.ObjectOperations.PATCH)
+        with reporter.step("Add APE rule with deny PATCH operation"):
+            frostfs_cli.ape_manager.add(
+                self.cluster.default_rpc_endpoint,
+                rule.chain_id,
+                target_name=container,
+                target_type="container",
+                rule=rule.as_string(),
+            )
+
+        with reporter.step("Wait for one block"):
+            self.wait_for_blocks(1)
+
+        with reporter.step("Check that PATCH is not allowed with owner wallet"):
+            patch_params["ranges"] = ["300:200"]
+            patch_params["new_attrs"] = "owner_2=false"
+
+            with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
+                grpc_client.object.patch(**patch_params)
+
+        with reporter.step("Check that PATCH is not allowed with another wallet"):
+            patch_params["ranges"] = ["100:50"]
+            patch_params["new_attrs"] = "other_2=false"
+
+            with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
+                grpc_client_with_other_wallet.object.patch(**patch_params)
+
+        with reporter.step("Check that PATCH is allowed with container wallet as rule is ignored"):
+            patch_params["ranges"] = ["600:0"]
+            patch_params["new_attrs"] = "container_2=true"
+
+            patched_oid = grpc_client_with_container_wallet.object.patch(**patch_params)
+            assert patched_oid != patch_params["oid"], "OID of patched object must be different from original one"
+            patch_params["oid"] = patched_oid
+
+        with reporter.step("Check that PATCH is not allowed with ir waller"):
+            patch_params["ranges"] = ["0:1000"]
+            patch_params["new_attrs"] = "ir_2=true"
+
+            with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
+                grpc_client_with_ir_wallet.object.patch(**patch_params)
+
+        with reporter.step("Remove APE rule"):
+            frostfs_cli.ape_manager.remove(
+                self.cluster.default_rpc_endpoint,
+                rule.chain_id,
+                target_name=container,
+                target_type="container",
+            )
+
+        with reporter.step("Wait for one block"):
+            self.wait_for_blocks(1)
+
+        with reporter.step("Check that PATCH is available with owner wallet"):
+            patch_params["ranges"] = ["300:200"]
+            patch_params["new_attrs"] = "owner_3=true"
+
+            patched_oid = grpc_client.object.patch(**patch_params)
+            assert patched_oid != patch_params["oid"], "OID of patched object must be different from original one"
+            patch_params["oid"] = patched_oid
+
+        with reporter.step("Check that PATCH is available with another wallet"):
+            patch_params["ranges"] = ["100:50"]
+            patch_params["new_attrs"] = "other_3=true"
+
+            patched_oid = grpc_client_with_other_wallet.object.patch(**patch_params)
+            assert patched_oid != patch_params["oid"], "OID of patched object must be different from original one"
+            patch_params["oid"] = patched_oid
+
+        with reporter.step("Check that PATCH is available with container wallet"):
+            patch_params["ranges"] = ["600:0"]
+            patch_params["new_attrs"] = "container_3=true"
+
+            patched_oid = grpc_client_with_container_wallet.object.patch(**patch_params)
+            assert patched_oid != patch_params["oid"], "OID of patched object must be different from original one"
+            patch_params["oid"] = patched_oid
+
+        with reporter.step("Check that PATCH is available with ir wallet"):
+            patch_params["ranges"] = ["0:1000"]
+            patch_params["new_attrs"] = "ir_3=true"
+
+            patched_oid = grpc_client_with_ir_wallet.object.patch(**patch_params)
+            assert patched_oid != patch_params["oid"], "OID of patched object must be different from original one"
+            patch_params["oid"] = patched_oid
+
+        attrs = {"owner", "other", "container", "ir", "container_2", "owner_3", "other_3", "container_3", "ir_3"}
+
+        with reporter.step("Ensure that all attributes match expected values"):
+            object_info: dict = grpc_client.object.head(container, patch_params["oid"], self.cluster.default_rpc_endpoint)
+            object_attrs: dict = object_info["header"]["attributes"]
+            assert attrs <= {k for k in object_attrs.keys()}, f"Received attributes do not match expected ones: {object_attrs}"
+            assert all(
+                v == "true" for k, v in object_attrs.items() if k in attrs
+            ), f"Received attributes do not match expected ones: {object_attrs}"

pytest_tests/testsuites/access/ape/test_bearer.py

@@ -2,8 +2,10 @@ import allure
 import pytest
 from frostfs_testlib import reporter
 from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
+from frostfs_testlib.resources.error_patterns import OBJECT_ACCESS_DENIED
 from frostfs_testlib.storage.dataclasses import ape
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
+from frostfs_testlib.storage.grpc_operations.interfaces import GrpcClientWrapper
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 from frostfs_testlib.utils.file_utils import TestFile
 
@@ -20,9 +22,13 @@ from ....helpers.container_access import (
 @pytest.mark.sanity
 @pytest.mark.bearer
 @pytest.mark.ape
+@pytest.mark.parametrize("user_tag", ["ApeBearer"], indirect=True)  # provide dedicated user with no APE side-policies
 class TestApeBearer(ClusterTestBase):
+    # TODO: Without PATCH operation,
+    # since it requires specific permissions that do not apply when testing all operations at once
     @allure.title("Operations with BearerToken (role={role}, obj_size={object_size})")
     @pytest.mark.parametrize("role", [ape.Role.OWNER, ape.Role.OTHERS], indirect=True)
+    @pytest.mark.parametrize("objects", [4], indirect=True)
     def test_bearer_token_operations(
         self,
         container: str,
@@ -31,11 +37,11 @@ class TestApeBearer(ClusterTestBase):
         temp_directory: str,
         test_wallet: WalletInfo,
         role: ape.Role,
-        file_path: TestFile,
+        test_file: TestFile,
         rpc_endpoint: str,
     ):
         with reporter.step(f"Check {role} has full access to container without bearer token"):
-            assert_full_access_to_container(test_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(test_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
         with reporter.step(f"Deny all operations for everyone via APE"):
             rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS)
@@ -54,10 +60,10 @@ class TestApeBearer(ClusterTestBase):
             )
 
         with reporter.step(f"Check {role} without token has no access to all operations with container"):
-            assert_no_access_to_container(test_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
+            assert_no_access_to_container(test_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
         with reporter.step(f"Check {role} with token has access to all operations with container"):
-            assert_full_access_to_container(test_wallet, container, objects.pop(), file_path, self.shell, self.cluster, bearer)
+            assert_full_access_to_container(test_wallet, container, objects.pop(), test_file, self.shell, self.cluster, bearer)
 
         with reporter.step(f"Remove deny rule from APE"):
             frostfs_cli.ape_manager.remove(rpc_endpoint, rule.chain_id, target_name=container, target_type="container")
@@ -66,9 +72,121 @@ class TestApeBearer(ClusterTestBase):
             self.wait_for_blocks()
 
         with reporter.step(f"Check {role} without token has access to all operations with container"):
-            assert_full_access_to_container(test_wallet, container, objects.pop(), file_path, self.shell, self.cluster)
+            assert_full_access_to_container(test_wallet, container, objects.pop(), test_file, self.shell, self.cluster)
+
+    # ^
+    @allure.title("Patch operation with BearerToken (object_size={object_size})")
+    @pytest.mark.parametrize("objects", [1], indirect=True)
+    def test_patch_object_with_bearer_token(
+        self,
+        frostfs_cli: FrostfsCli,
+        grpc_client_with_other_wallet: GrpcClientWrapper,
+        container: str,
+        objects: list[str],
+        test_file: TestFile,
+        temp_directory: str,
+    ):
+        oid = objects[0]
+
+        with reporter.step("Check if the patch is available with another wallet"):
+            patched_oid = grpc_client_with_other_wallet.object.patch(
+                container,
+                oid,
+                self.cluster.default_rpc_endpoint,
+                ranges=["100:300"],
+                payloads=[test_file],
+                new_attrs="allow-patch=true",
+                timeout="200s",
+            )
+            assert patched_oid != oid, "OID of patched object must be different from original one"
+            oid = patched_oid
+
+        rule = ape.Rule(ape.Verb.DENY, ape.ObjectOperations.PATCH)
+
+        with reporter.step("Deny PATCH operation for everyone via APE"):
+            frostfs_cli.ape_manager.add(
+                self.cluster.default_rpc_endpoint,
+                rule.chain_id,
+                target_name=container,
+                target_type="container",
+                rule=rule.as_string(),
+            )
+
+        with reporter.step("Wait for one block"):
+            self.wait_for_blocks(1)
+
+        with reporter.step("Check that patch is not allowed with another wallet"):
+            with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
+                grpc_client_with_other_wallet.object.patch(
+                    container,
+                    oid,
+                    self.cluster.default_rpc_endpoint,
+                    ranges=["100:300"],
+                    payloads=[test_file],
+                    new_attrs="deny-patch=true",
+                    timeout="200s",
+                )
+
+        with reporter.step("Create bearer token with all operations allowed"):
+            bearer = create_bearer_token(
+                frostfs_cli,
+                temp_directory,
+                container,
+                rule=ape.Rule(ape.Verb.ALLOW, ALL_OBJECT_OPERATIONS),
+                endpoint=self.cluster.default_rpc_endpoint,
+            )
+
+        with reporter.step("Check that patch is available with another wallet with BearerToken"):
+            patched_oid = grpc_client_with_other_wallet.object.patch(
+                container,
+                oid,
+                self.cluster.default_rpc_endpoint,
+                bearer=bearer,
+                ranges=["100:300"],
+                payloads=[test_file],
+                new_attrs="bearer-patch=true",
+                timeout="200s",
+            )
+            assert patched_oid != oid, "OID of patched object must be different from original one"
+            oid = patched_oid
+
+        with reporter.step(f"Remove deny rule from APE"):
+            frostfs_cli.ape_manager.remove(
+                self.cluster.default_rpc_endpoint,
+                rule.chain_id,
+                target_name=container,
+                target_type="container",
+            )
+
+        with reporter.step("Wait for one block"):
+            self.wait_for_blocks(1)
+
+        with reporter.step("Check if the patch is available with another wallet"):
+            patched_oid = grpc_client_with_other_wallet.object.patch(
+                container,
+                oid,
+                self.cluster.default_rpc_endpoint,
+                bearer=bearer,
+                ranges=["100:300"],
+                payloads=[test_file],
+                new_attrs="allow-patch-2=true",
+                timeout="200s",
+            )
+            assert patched_oid != oid, "OID of patched object must be different from original one"
+            oid = patched_oid
+
+        attrs = {"allow-patch", "bearer-patch", "allow-patch-2"}
+
+        with reporter.step("Ensure that all attributes match expected values"):
+            object_info: dict = grpc_client_with_other_wallet.object.head(container, oid, self.cluster.default_rpc_endpoint)
+            object_attrs: dict = object_info["header"]["attributes"]
+            assert attrs <= {k for k in object_attrs.keys()}, f"Received attributes do not match expected ones: {object_attrs}"
+            assert all(
+                v == "true" for k, v in object_attrs.items() if k in attrs
+            ), f"Received attributes do not match expected ones: {object_attrs}"
 
     @allure.title("BearerToken for compound operations (obj_size={object_size})")
+    @pytest.mark.parametrize("objects", [4], indirect=True)
     def test_bearer_token_compound_operations(
         self,
         frostfs_cli: FrostfsCli,
@@ -78,7 +196,7 @@ class TestApeBearer(ClusterTestBase):
         container: str,
         objects: list[str],
         rpc_endpoint: str,
-        file_path: TestFile,
+        test_file: TestFile,
     ):
         """
         Bearer Token COMPLETLY overrides chains set for the specific target.
@@ -150,7 +268,7 @@ class TestApeBearer(ClusterTestBase):
                 ape.ObjectOperations.PUT,
                 ape.ObjectOperations.HEAD,
                 ape.ObjectOperations.GET_RANGE,
-                # Delete also requires PUT (to make tobstone) and HEAD (to get simple objects header)
+                # Delete also requires PUT (to make tombstone) and HEAD (to get simple objects header)
                 ape.ObjectOperations.DELETE,
             ],
             ape.Role.OTHERS: [
@@ -179,7 +297,7 @@ class TestApeBearer(ClusterTestBase):
 
         for role, wallet in wallets_map.items():
             with reporter.step(f"Assert access to container without bearer token for {role}"):
-                assert_access_to_container(access_map[role], wallet, container, objects.pop(), file_path, self.shell, self.cluster)
+                assert_access_to_container(access_map[role], wallet, container, objects.pop(), test_file, self.shell, self.cluster)
 
         bearer_tokens = {}
         for role in wallets_map.keys():
@@ -191,5 +309,5 @@ class TestApeBearer(ClusterTestBase):
         for role, wallet in wallets_map.items():
             with reporter.step(f"Assert access to container with bearer token for {role}"):
                 assert_access_to_container(
-                    bt_access_map[role], wallet, container, objects.pop(), file_path, self.shell, self.cluster, bearer_tokens[role]
+                    bt_access_map[role], wallet, container, objects.pop(), test_file, self.shell, self.cluster, bearer_tokens[role]
                 )

pytest_tests/testsuites/access/conftest.py

@@ -2,16 +2,18 @@ import json
 
 import pytest
 from frostfs_testlib import reporter
+from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
+from frostfs_testlib.resources.cli import FROSTFS_CLI_EXEC
 from frostfs_testlib.shell import Shell
 from frostfs_testlib.steps.cli.container import search_nodes_with_container
 from frostfs_testlib.steps.cli.object import put_object_to_random_node
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode
 from frostfs_testlib.storage.dataclasses import ape
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
+from frostfs_testlib.storage.grpc_operations.client_wrappers import CliClientWrapper
+from frostfs_testlib.storage.grpc_operations.interfaces import GrpcClientWrapper
 from frostfs_testlib.testing.parallel import parallel
 
-OBJECT_COUNT = 5
-
 
 @pytest.fixture(scope="session")
 def ir_wallet(cluster: Cluster) -> WalletInfo:
@@ -40,13 +42,22 @@ def test_wallet(default_wallet: WalletInfo, other_wallet: WalletInfo, role: ape.
     return role_to_wallet_map[role]
 
 
-@pytest.fixture
-def objects(container: str, default_wallet: WalletInfo, client_shell: Shell, cluster: Cluster, file_path: str):
+@pytest.fixture(scope="function", params=[5])
+def objects(
+    container: str,
+    default_wallet: WalletInfo,
+    client_shell: Shell,
+    cluster: Cluster,
+    test_file: str,
+    request: pytest.FixtureRequest,
+):
+    object_count = request.param
+
     with reporter.step("Add test objects to container"):
         put_results = parallel(
-            [put_object_to_random_node] * OBJECT_COUNT,
+            [put_object_to_random_node] * object_count,
             wallet=default_wallet,
-            path=file_path,
+            path=test_file,
             cid=container,
             shell=client_shell,
             cluster=cluster,
@@ -70,3 +81,18 @@ def container_nodes(default_wallet: WalletInfo, container: str, client_shell: Sh
 @pytest.fixture
 def container_node_wallet(container_nodes: list[ClusterNode]) -> WalletInfo:
     return WalletInfo.from_node(container_nodes[0].storage_node)
+
+
+@pytest.fixture
+def grpc_client_with_container_wallet(client_shell: Shell, container_node_wallet: WalletInfo) -> GrpcClientWrapper:
+    return CliClientWrapper(FrostfsCli(client_shell, FROSTFS_CLI_EXEC, container_node_wallet.config_path))
+
+
+@pytest.fixture(scope="session")
+def grpc_client_with_other_wallet(client_shell: Shell, other_wallet: WalletInfo) -> GrpcClientWrapper:
+    return CliClientWrapper(FrostfsCli(client_shell, FROSTFS_CLI_EXEC, other_wallet.config_path))
+
+
+@pytest.fixture(scope="session")
+def grpc_client_with_ir_wallet(client_shell: Shell, ir_wallet: WalletInfo) -> GrpcClientWrapper:
+    return CliClientWrapper(FrostfsCli(client_shell, FROSTFS_CLI_EXEC, ir_wallet.config_path))

pytest_tests/testsuites/ape/test_ape_local_object_allow.py

@@ -3,12 +3,10 @@ import pytest
 from frostfs_testlib import reporter
 from frostfs_testlib.cli import FrostfsCli
 from frostfs_testlib.resources.error_patterns import NO_RULE_FOUND_OBJECT
-from frostfs_testlib.steps.cli.object import delete_object, get_object, get_range, get_range_hash, head_object, put_object, search_object
-from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
-from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
+from frostfs_testlib.storage.grpc_operations.interfaces import GrpcClientWrapper
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 from frostfs_testlib.testing.test_control import expect_not_raises
-from frostfs_testlib.utils.file_utils import generate_file
+from frostfs_testlib.utils.file_utils import TestFile
 
 from ...helpers.container_request import ContainerRequest
 
@@ -20,12 +18,13 @@ REP1_MSK = ContainerRequest("REP 1 IN MOW CBF 1 SELECT 1 FROM MSK AS MOW FILTER
 @pytest.mark.ape_object
 @pytest.mark.ape_allow
 @pytest.mark.parametrize("container_request", [REP1_MSK], indirect=True)
+@pytest.mark.parametrize("user_tag", ["ApeLocalOverrideAllow"], indirect=True)  # provide dedicated user with no APE side-policies
 class TestApeLocalOverrideAllow(ClusterTestBase):
     @allure.title("LocalOverride: Allow to GetObject in root tenant")
     def test_local_override_allow_to_get_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
+        grpc_client: GrpcClientWrapper,
         container: str,
         object_id: str,
     ):
@@ -40,11 +39,11 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
 
         with reporter.step("Check get object in container on the first node, expected allow"):
             with expect_not_raises():
-                get_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.get(container, object_id, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check get object in container on the second node, epxected access denied error"):
             with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
-                get_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.get(container, object_id, self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -55,15 +54,14 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
             )
 
     @allure.title("LocalOverride: Allow to PutObject in root tenant")
+    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
     def test_local_override_allow_to_put_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
-        simple_object_size: ObjectSize,
+        grpc_client: GrpcClientWrapper,
         container: str,
+        test_file: TestFile,
     ):
-        test_file = generate_file(simple_object_size.value)
-
         with reporter.step("Create local override on first node"):
             frostfs_cli_on_first_node.control.add_rule(
                 endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
@@ -75,11 +73,11 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
 
         with reporter.step("Check put object in container on the first node, expected allow"):
             with expect_not_raises():
-                put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.put(test_file, container, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
-        with reporter.step("Check get object in container on the second node, epxected access denied error"):
+        with reporter.step("Check put object in container on the second node, epxected access denied error"):
             with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
-                put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.put(test_file, container, self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -92,8 +90,8 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
     @allure.title("LocalOverride: Allow to HeadObject in root tenant")
     def test_local_override_allow_to_head_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
+        grpc_client: GrpcClientWrapper,
         container: str,
         object_id: str,
     ):
@@ -108,11 +106,11 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
 
         with reporter.step("Check head object in container on the first node, expected allow"):
             with expect_not_raises():
-                head_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.head(container, object_id, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check head object in container on the second node, expected access denied error"):
             with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
-                head_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.head(container, object_id, self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -125,8 +123,8 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
     @allure.title("LocalOverride: Allow to SearchObject in root tenant")
     def test_local_override_allow_to_search_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
+        grpc_client: GrpcClientWrapper,
         container: str,
     ):
         with reporter.step("Create local override on first node"):
@@ -140,11 +138,11 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
 
         with reporter.step("Check search object in container on the first node, expected allow"):
             with expect_not_raises():
-                search_object(default_wallet, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.search(container, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check search object from container on the second node, expected access denied error"):
             with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
-                search_object(default_wallet, container, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.search(container, self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -157,8 +155,8 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
     @allure.title("LocalOverride: Allow to RangeObject in root tenant")
     def test_local_override_allow_to_range_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
+        grpc_client: GrpcClientWrapper,
         container: str,
         object_id: str,
     ):
@@ -173,11 +171,11 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
 
         with reporter.step("Check get range object in container on the first node, expected allow"):
             with expect_not_raises():
-                get_range(default_wallet, container, object_id, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.range(container, object_id, "0:10", self.cluster.storage_nodes[0].get_rpc_endpoint())
 
-        with reporter.step("Check range object in container on the second node. expected access denied error"):
+        with reporter.step("Check get range object in container on the second node, expected access denied error"):
             with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
-                get_range(default_wallet, container, object_id, "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.range(container, object_id, "0:10", self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -190,8 +188,8 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
     @allure.title("LocalOverride: Allow to HashObject in root tenant")
     def test_local_override_allow_to_hash_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
+        grpc_client: GrpcClientWrapper,
         container: str,
         object_id: str,
     ):
@@ -206,11 +204,11 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
 
         with reporter.step("Check get range hash object in container on the first node, expected allow"):
             with expect_not_raises():
-                get_range_hash(default_wallet, container, object_id, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.hash(self.cluster.storage_nodes[0].get_rpc_endpoint(), container, object_id, range="0:10")
 
         with reporter.step("Check get range hash object in container on the second node, expected access denied error"):
             with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
-                get_range_hash(default_wallet, container, object_id, "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.hash(self.cluster.storage_nodes[1].get_rpc_endpoint(), container, object_id, range="0:10")
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -223,8 +221,8 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
     @allure.title("LocalOverride: Allow to DeleteObject in root tenant")
     def test_local_override_allow_to_delete_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
+        grpc_client: GrpcClientWrapper,
         container: str,
         object_id: str,
     ):
@@ -239,11 +237,11 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
 
         with reporter.step("Check delete object from container on the second node, expected access denied error"):
             with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
-                delete_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.delete(container, object_id, self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Check delete object in container on the first node, expected allow"):
             with expect_not_raises():
-                delete_object(default_wallet, container, object_id, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.delete(container, object_id, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -252,3 +250,55 @@ class TestApeLocalOverrideAllow(ClusterTestBase):
                 target_name=container,
                 chain_id="allowDeleteObject",
             )
+
+    @allure.title("LocalOverride: Allow to PatchObject in root tenant")
+    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
+    def test_local_override_allow_to_patch_object_root(
+        self,
+        frostfs_cli_on_first_node: FrostfsCli,
+        grpc_client: GrpcClientWrapper,
+        container: str,
+        object_id: str,
+        test_file: TestFile,
+    ):
+        with reporter.step("Create local override on first node"):
+            frostfs_cli_on_first_node.control.add_rule(
+                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+                target_type="container",
+                target_name=container,
+                chain_id="allowPatchObject",
+                rule=f"allow Object.Patch *",
+            )
+
+        with reporter.step("Check patch object in container on the second node, epxected access denied error"):
+            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
+                grpc_client.object.patch(
+                    container,
+                    object_id,
+                    self.cluster.storage_nodes[1].get_rpc_endpoint(),
+                    ranges=["500:300"],
+                    payloads=[test_file],
+                    new_attrs="patched=false",
+                    timeout="200s",
+                )
+
+        with reporter.step("Check patch object in container on the first node, expected allow"):
+            with expect_not_raises():
+                patched_oid = grpc_client.object.patch(
+                    container,
+                    object_id,
+                    self.cluster.storage_nodes[0].get_rpc_endpoint(),
+                    ranges=["100:200"],
+                    payloads=[test_file],
+                    new_attrs="patched=true",
+                    timeout="200s",
+                )
+                assert patched_oid != object_id, "OID of patched object must be different from original one"
+
+        with reporter.step("Delete a rule"):
+            frostfs_cli_on_first_node.control.remove_rule(
+                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+                target_type="container",
+                target_name=container,
+                chain_id="allowPatchObject",
+            )

pytest_tests/testsuites/ape/test_ape_local_object_deny.py

@@ -1,20 +1,15 @@
 import allure
 import pytest
+from frostfs_testlib import reporter
 from frostfs_testlib.cli import FrostfsCli
-from frostfs_testlib.reporter import get_reporter
-from frostfs_testlib.resources.error_patterns import OBJECT_ACCESS_DENIED, RULE_ACCESS_DENIED_OBJECT
-from frostfs_testlib.steps.cli.object import delete_object, get_object, get_range, get_range_hash, head_object, put_object, search_object
-from frostfs_testlib.storage.dataclasses.ape import Operations
-from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
-from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
+from frostfs_testlib.resources.error_patterns import RULE_ACCESS_DENIED_OBJECT
+from frostfs_testlib.storage.grpc_operations.interfaces import GrpcClientWrapper
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 from frostfs_testlib.testing.test_control import expect_not_raises
-from frostfs_testlib.utils.file_utils import generate_file
+from frostfs_testlib.utils.file_utils import TestFile
 
 from ...helpers.container_request import APE_EVERYONE_ALLOW_ALL, ContainerRequest
 
-reporter = get_reporter()
-
 REP2 = ContainerRequest("REP 2", ape_rules=APE_EVERYONE_ALLOW_ALL, short_name="REP2_allow_all_ape")
 
 
@@ -25,15 +20,14 @@ REP2 = ContainerRequest("REP 2", ape_rules=APE_EVERYONE_ALLOW_ALL, short_name="R
 class TestApeLocalOverrideDeny(ClusterTestBase):
     @allure.title("LocalOverride: Deny to GetObject in root tenant")
     @pytest.mark.parametrize("container_request", [REP2], indirect=True)
+    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
     def test_local_override_deny_to_get_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
-        simple_object_size: ObjectSize,
+        grpc_client: GrpcClientWrapper,
         container: str,
+        test_file: TestFile,
     ):
-        test_file = generate_file(simple_object_size.value)
-
         with reporter.step("Create local override on first node"):
             frostfs_cli_on_first_node.control.add_rule(
                 endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
@@ -44,15 +38,15 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
             )
 
         with reporter.step("Put object in container on the first node"):
-            oid = put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+            oid = grpc_client.object.put(test_file, container, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check get object from container on the first node, expected access denied error"):
             with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT):
-                get_object(default_wallet, container, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.get(container, oid, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check get object from container on the second node, expected allow"):
             with expect_not_raises():
-                get_object(default_wallet, container, oid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.get(container, oid, self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -64,19 +58,18 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
 
         with reporter.step("Check get object in container on the first node, expected allow"):
             with expect_not_raises():
-                get_object(default_wallet, container, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.get(container, oid, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
     @allure.title("LocalOverride: Deny to PutObject in root tenant")
     @pytest.mark.parametrize("container_request", [REP2], indirect=True)
+    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
     def test_local_override_deny_to_put_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
-        simple_object_size: ObjectSize,
+        grpc_client: GrpcClientWrapper,
         container: str,
+        test_file: TestFile,
     ):
-        test_file = generate_file(simple_object_size.value)
-
         with reporter.step("Create local override on first node"):
             frostfs_cli_on_first_node.control.add_rule(
                 endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
@@ -87,14 +80,12 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
             )
 
         with reporter.step("Check put object from container on the first node, expected access denied error"):
-            with pytest.raises(RuntimeError, match=OBJECT_ACCESS_DENIED):
-                put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+            with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT):
+                grpc_client.object.put(test_file, container, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check put object from container on the second node, expected allow"):
             with expect_not_raises():
-                put_object(
-                    default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint(), copies_number=3
-                )
+                grpc_client.object.put(test_file, container, self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -106,19 +97,18 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
 
         with reporter.step("Check get object in container on the first node, expected allow"):
             with expect_not_raises():
-                put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.put(test_file, container, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
     @allure.title("LocalOverride: Deny to HeadObject in root tenant")
     @pytest.mark.parametrize("container_request", [REP2], indirect=True)
+    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
     def test_local_override_deny_to_head_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
-        simple_object_size: ObjectSize,
+        grpc_client: GrpcClientWrapper,
         container: str,
+        test_file: TestFile,
     ):
-        test_file = generate_file(simple_object_size.value)
-
         with reporter.step("Create local override on first node"):
             frostfs_cli_on_first_node.control.add_rule(
                 endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
@@ -129,15 +119,15 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
             )
 
         with reporter.step("Put object in container on the first node"):
-            oid = put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+            oid = grpc_client.object.put(test_file, container, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check head object from container on the first node, expected access denied error"):
             with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT):
-                head_object(default_wallet, container, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.head(container, oid, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check head object from container on the second node, expected allow"):
             with expect_not_raises():
-                head_object(default_wallet, container, oid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.head(container, oid, self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -149,14 +139,14 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
 
         with reporter.step("Check head object in container on the first node, expected allow"):
             with expect_not_raises():
-                head_object(default_wallet, container, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.head(container, oid, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
     @allure.title("LocalOverride: Deny to SearchObject in root tenant")
     @pytest.mark.parametrize("container_request", [REP2], indirect=True)
     def test_local_override_deny_to_search_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
+        grpc_client: GrpcClientWrapper,
         container: str,
     ):
         with reporter.step("Create local override on first node"):
@@ -169,12 +159,12 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
             )
 
         with reporter.step("Check search object from container on the first node, expected access denied error"):
-            with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.SEARCH_OBJECT)):
-                search_object(default_wallet, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+            with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT):
+                grpc_client.object.search(container, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check search object from container on the second node, expected allow"):
             with expect_not_raises():
-                search_object(default_wallet, container, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.search(container, self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -186,19 +176,18 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
 
         with reporter.step("Check search object in container on the first node, expected allow"):
             with expect_not_raises():
-                search_object(default_wallet, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.search(container, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
     @allure.title("LocalOverride: Deny to RangeObject in root tenant")
     @pytest.mark.parametrize("container_request", [REP2], indirect=True)
+    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
     def test_local_override_deny_to_range_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
-        simple_object_size: ObjectSize,
+        grpc_client: GrpcClientWrapper,
         container: str,
+        test_file: TestFile,
     ):
-        test_file = generate_file(simple_object_size.value)
-
         with reporter.step("Create local override on first node"):
             frostfs_cli_on_first_node.control.add_rule(
                 endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
@@ -209,15 +198,15 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
             )
 
         with reporter.step("Put object in container on the first node"):
-            oid = put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+            oid = grpc_client.object.put(test_file, container, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check range object from container on the first node, expected access denied error"):
-            with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.RANGE_OBJECT)):
-                get_range(default_wallet, container, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+            with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT):
+                grpc_client.object.range(container, oid, "0:10", self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check get range object from container on the second node, expected allow"):
             with expect_not_raises():
-                get_range(default_wallet, container, oid, "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.range(container, oid, "0:10", self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -229,19 +218,18 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
 
         with reporter.step("Check get range object in container on the first node, expected allow"):
             with expect_not_raises():
-                get_range(default_wallet, container, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.range(container, oid, "0:10", self.cluster.storage_nodes[0].get_rpc_endpoint())
 
     @allure.title("LocalOverride: Deny to HashObject in root tenant")
     @pytest.mark.parametrize("container_request", [REP2], indirect=True)
+    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
     def test_local_override_deny_to_hash_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
-        simple_object_size: ObjectSize,
+        grpc_client: GrpcClientWrapper,
         container: str,
+        test_file: TestFile,
     ):
-        test_file = generate_file(simple_object_size.value)
-
         with reporter.step("Create local override on first node"):
             frostfs_cli_on_first_node.control.add_rule(
                 endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
@@ -252,15 +240,15 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
             )
 
         with reporter.step("Put object in container on the first node"):
-            oid = put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+            oid = grpc_client.object.put(test_file, container, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check get range hash object from container on the first node, expected access denied error"):
-            with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.HASH_OBJECT)):
-                get_range_hash(default_wallet, container, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+            with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT):
+                grpc_client.object.hash(self.cluster.storage_nodes[0].get_rpc_endpoint(), container, oid, range="0:10")
 
         with reporter.step("Check get range hash object from container on the second node, expected allow"):
             with expect_not_raises():
-                get_range_hash(default_wallet, container, oid, "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.hash(self.cluster.storage_nodes[1].get_rpc_endpoint(), container, oid, range="0:10")
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -272,19 +260,18 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
 
         with reporter.step("Check get range hash object in container on the first node, expected allow"):
             with expect_not_raises():
-                get_range_hash(default_wallet, container, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.hash(self.cluster.storage_nodes[0].get_rpc_endpoint(), container, oid, range="0:10")
 
     @allure.title("LocalOverride: Deny to DeleteObject in root tenant")
     @pytest.mark.parametrize("container_request", [REP2], indirect=True)
+    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
     def test_local_override_deny_to_delete_object_root(
         self,
-        default_wallet: WalletInfo,
         frostfs_cli_on_first_node: FrostfsCli,
-        simple_object_size: ObjectSize,
+        grpc_client: GrpcClientWrapper,
         container: str,
+        test_file: TestFile,
     ):
-        test_file = generate_file(simple_object_size.value)
-
         with reporter.step("Create local override on first node"):
             frostfs_cli_on_first_node.control.add_rule(
                 endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
@@ -295,30 +282,26 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
             )
 
         with reporter.step("Put objects in container on the first node"):
-            oid_1 = put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
-            oid_2 = put_object(default_wallet, test_file, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+            oid_1 = grpc_client.object.put(test_file, container, self.cluster.storage_nodes[0].get_rpc_endpoint())
+            oid_2 = grpc_client.object.put(test_file, container, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Search object in container on the first node"):
-            search_object_in_container_1 = search_object(
-                default_wallet, container, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()
-            )
+            search_object_in_container_1 = grpc_client.object.search(container, self.cluster.storage_nodes[0].get_rpc_endpoint())
             assert oid_1 in search_object_in_container_1, f"Object {oid_1} was not found"
             assert oid_2 in search_object_in_container_1, f"Object {oid_2} was not found"
 
         with reporter.step("Search object from container on the second node"):
-            search_object_in_container_2 = search_object(
-                default_wallet, container, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint()
-            )
+            search_object_in_container_2 = grpc_client.object.search(container, self.cluster.storage_nodes[1].get_rpc_endpoint())
             assert oid_1 in search_object_in_container_2, f"Object {oid_1} was not found"
             assert oid_2 in search_object_in_container_2, f"Object {oid_2} was not found"
 
         with reporter.step("Check delete object from container on the first node, expected access denied error"):
             with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT):
-                delete_object(default_wallet, container, oid_1, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.delete(container, oid_1, self.cluster.storage_nodes[0].get_rpc_endpoint())
 
         with reporter.step("Check delete object from container on the second node, expected allow"):
             with expect_not_raises():
-                delete_object(default_wallet, container, oid_2, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+                grpc_client.object.delete(container, oid_2, self.cluster.storage_nodes[1].get_rpc_endpoint())
 
         with reporter.step("Delete a rule"):
             frostfs_cli_on_first_node.control.remove_rule(
@@ -330,4 +313,70 @@ class TestApeLocalOverrideDeny(ClusterTestBase):
 
         with reporter.step("Check delete object in container on the first node, expected allow"):
             with expect_not_raises():
-                delete_object(default_wallet, container, oid_1, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+                grpc_client.object.delete(container, oid_1, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+    @allure.title("LocalOverride: Deny to PatchObject in root tenant")
+    @pytest.mark.parametrize("container_request", [REP2], indirect=True)
+    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
+    def test_local_override_deny_to_patch_object_root(
+        self,
+        frostfs_cli_on_first_node: FrostfsCli,
+        grpc_client: GrpcClientWrapper,
+        test_file: TestFile,
+        container: str,
+        object_id: str,
+    ):
+        with reporter.step("Create local override on first node"):
+            frostfs_cli_on_first_node.control.add_rule(
+                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+                target_type="container",
+                target_name=container,
+                chain_id="denyPatchObject",
+                rule=f"deny Object.Patch /{container}/*",
+            )
+
+        with reporter.step("Check patch object from container on the first node, expected access denied error"):
+            with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT):
+                grpc_client.object.patch(
+                    container,
+                    object_id,
+                    self.cluster.storage_nodes[0].get_rpc_endpoint(),
+                    ranges=["0:350"],
+                    payloads=[test_file],
+                    new_attrs="patched_by_first_node=false",
+                    timeout="200s",
+                )
+
+        with reporter.step("Check patch object from container on the second node, expected allow"):
+            with expect_not_raises():
+                patched_oid_1 = grpc_client.object.patch(
+                    container,
+                    object_id,
+                    self.cluster.storage_nodes[1].get_rpc_endpoint(),
+                    ranges=["200:400"],
+                    payloads=[test_file],
+                    new_attrs="patched_by_second_node=true",
+                    timeout="200s",
+                )
+                assert patched_oid_1 != object_id, "OID of patched object must be different from original one"
+
+        with reporter.step("Delete a rule"):
+            frostfs_cli_on_first_node.control.remove_rule(
+                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+                target_type="container",
+                target_name=container,
+                chain_id="denyPatchObject",
+            )
+
+        with reporter.step("Check patch object in container on the first node, expected allow"):
+            with expect_not_raises():
+                patched_oid_2 = grpc_client.object.patch(
+                    container,
+                    patched_oid_1,
+                    self.cluster.storage_nodes[0].get_rpc_endpoint(),
+                    ranges=["600:0"],
+                    payloads=[test_file],
+                    new_attrs="patched_by_first_node=true",
+                    timeout="200s",
+                )
+                assert patched_oid_1 != patched_oid_2, "OID of patched object must be different from original one"

pytest_tests/testsuites/conftest.py

@@ -8,18 +8,18 @@ import pytest
 from dateutil import parser
 from frostfs_testlib import plugins, reporter
 from frostfs_testlib.cli import FrostfsCli
+from frostfs_testlib.clients import AwsCliClient, Boto3ClientWrapper, S3ClientWrapper, S3HttpClient
+from frostfs_testlib.clients.s3 import BucketContainerResolver, VersioningStatus
 from frostfs_testlib.credentials.interfaces import CredentialsProvider, User
 from frostfs_testlib.healthcheck.interfaces import Healthcheck
 from frostfs_testlib.hosting import Hosting
 from frostfs_testlib.resources import optionals
 from frostfs_testlib.resources.common import COMPLEX_OBJECT_CHUNKS_COUNT, COMPLEX_OBJECT_TAIL_SIZE, SIMPLE_OBJECT_SIZE
-from frostfs_testlib.s3 import AwsCliClient, Boto3ClientWrapper, S3ClientWrapper, VersioningStatus
-from frostfs_testlib.s3.interfaces import BucketContainerResolver
 from frostfs_testlib.shell import LocalShell, Shell
+from frostfs_testlib.steps import s3_helper
 from frostfs_testlib.steps.cli.container import DEFAULT_EC_PLACEMENT_RULE, DEFAULT_PLACEMENT_RULE, FROSTFS_CLI_EXEC
 from frostfs_testlib.steps.cli.object import get_netmap_netinfo
 from frostfs_testlib.steps.epoch import ensure_fresh_epoch
-from frostfs_testlib.steps.s3 import s3_helper
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode
 from frostfs_testlib.storage.controllers.cluster_state_controller import ClusterStateController
 from frostfs_testlib.storage.dataclasses.frostfs_services import StorageNode
@@ -206,8 +206,8 @@ def ec_placement_policy() -> PlacementPolicy:
 
 @pytest.fixture(scope="session")
 @allure.title("Init Frostfs CLI")
-def frostfs_cli(client_shell: Shell, default_wallet: WalletInfo) -> FrostfsCli:
-    return FrostfsCli(client_shell, FROSTFS_CLI_EXEC, default_wallet.config_path)
+def frostfs_cli(client_shell: Shell, wallet: WalletInfo) -> FrostfsCli:
+    return FrostfsCli(client_shell, FROSTFS_CLI_EXEC, wallet.config_path)
 
 
 @pytest.fixture(scope="session")
@@ -287,20 +287,34 @@ def credentials_provider(cluster: Cluster) -> CredentialsProvider:
     ],
 )
 def s3_client(
-    default_user: User,
+    user: User,
     s3_policy: Optional[str],
     cluster: Cluster,
     request: pytest.FixtureRequest,
     credentials_provider: CredentialsProvider,
 ) -> S3ClientWrapper:
     node = cluster.cluster_nodes[0]
-    credentials_provider.S3.provide(default_user, node, s3_policy)
+    credentials_provider.S3.provide(user, node, s3_policy)
 
     s3_client_cls = request.param
-    client = s3_client_cls(default_user.s3_credentials.access_key, default_user.s3_credentials.secret_key, cluster.default_s3_gate_endpoint)
+    client = s3_client_cls(user.s3_credentials.access_key, user.s3_credentials.secret_key, cluster.default_s3_gate_endpoint)
     return client
 
 
+@allure.title("[Session] Create S3 http client")
+@pytest.fixture(scope="session")
+def s3_http_client(
+    default_user: User, s3_policy: Optional[str], cluster: Cluster, credentials_provider: CredentialsProvider
+) -> S3HttpClient:
+    node = cluster.cluster_nodes[0]
+    credentials_provider.S3.provide(default_user, node, s3_policy)
+    return S3HttpClient(
+        cluster.default_s3_gate_endpoint,
+        default_user.s3_credentials.access_key,
+        default_user.s3_credentials.secret_key,
+    )
+
+
 @pytest.fixture
 def versioning_status(request: pytest.FixtureRequest) -> VersioningStatus:
     if "param" in request.__dict__:
@@ -443,19 +457,50 @@ def default_user(credentials_provider: CredentialsProvider, cluster: Cluster) ->
     return user
 
 
-@reporter.step("Get wallet for default user")
 @pytest.fixture(scope="session")
-def default_wallet(default_user: User) -> WalletInfo:
-    return default_user.wallet
+@cached_fixture(optionals.OPTIONAL_CACHE_FIXTURES)
+def users_pool(credentials_provider: CredentialsProvider, cluster: Cluster) -> list[User]:
+    users = [User(string_utils.unique_name("user-")) for _ in range(WALLTETS_IN_POOL)]
+    parallel(credentials_provider.GRPC.provide, users, cluster_node=cluster.cluster_nodes[0])
+    return users
+
+
+@pytest.fixture(scope="session")
+def user_tag(request: pytest.FixtureRequest) -> str:
+    tag = "default"
+    if "param" in request.__dict__:
+        tag = request.param
+
+    return tag
 
 
 @pytest.fixture(scope="session")
 @cached_fixture(optionals.OPTIONAL_CACHE_FIXTURES)
-def wallets_pool(credentials_provider: CredentialsProvider, cluster: Cluster) -> list[WalletInfo]:
-    users = [User(string_utils.unique_name("user-")) for _ in range(WALLTETS_IN_POOL)]
-    parallel(credentials_provider.GRPC.provide, users, cluster_node=cluster.cluster_nodes[0])
+@reporter.step("Create {user_tag} user")
+def user(user_tag: str) -> User:
+    user = User(string_utils.unique_name("user-"))
+    user.attributes["tag"] = user_tag
 
-    return [user.wallet for user in users]
+    return user
+
+
+@pytest.fixture(scope="session")
+def wallet(user: User, credentials_provider: CredentialsProvider, cluster: Cluster) -> WalletInfo:
+    credentials_provider.GRPC.provide(user, cluster.cluster_nodes[0])
+    return user.wallet
+
+
+# TODO: Migrate tests to fixture wallet above
+@reporter.step("Get wallet for default user")
+@pytest.fixture(scope="session")
+def default_wallet(wallet: WalletInfo) -> WalletInfo:
+    return wallet
+
+
+@pytest.fixture(scope="session")
+@cached_fixture(optionals.OPTIONAL_CACHE_FIXTURES)
+def wallets_pool(users_pool: list[User]) -> list[WalletInfo]:
+    return [user.wallet for user in users_pool]
 
 
 @pytest.fixture(scope="session")
@@ -524,26 +569,26 @@ def multiple_containers_request(request: pytest.FixtureRequest) -> ContainerRequ
 
 @pytest.fixture
 def container(
-    default_wallet: WalletInfo,
+    wallet: WalletInfo,
     frostfs_cli: FrostfsCli,
     client_shell: Shell,
     cluster: Cluster,
     rpc_endpoint: str,
     container_request: ContainerRequest,
 ) -> str:
-    return create_container_with_ape(container_request, frostfs_cli, default_wallet, client_shell, cluster, rpc_endpoint)
+    return create_container_with_ape(container_request, frostfs_cli, wallet, client_shell, cluster, rpc_endpoint)
 
 
 @pytest.fixture
 def containers(
-    default_wallet: WalletInfo,
+    wallet: WalletInfo,
     frostfs_cli: FrostfsCli,
     client_shell: Shell,
     cluster: Cluster,
     rpc_endpoint: str,
     multiple_containers_request: MultipleContainersRequest,
 ) -> list[str]:
-    return create_containers_with_ape(frostfs_cli, default_wallet, client_shell, cluster, rpc_endpoint, multiple_containers_request)
+    return create_containers_with_ape(frostfs_cli, wallet, client_shell, cluster, rpc_endpoint, multiple_containers_request)
 
 
 @pytest.fixture()

pytest_tests/testsuites/failovers/test_failover_storage.py

@@ -6,9 +6,9 @@ from time import sleep
 import allure
 import pytest
 from frostfs_testlib import reporter
+from frostfs_testlib.clients.s3 import BucketContainerResolver, S3ClientWrapper, VersioningStatus
 from frostfs_testlib.resources.common import MORPH_BLOCK_TIME
-from frostfs_testlib.s3 import S3ClientWrapper, VersioningStatus
-from frostfs_testlib.s3.interfaces import BucketContainerResolver
+from frostfs_testlib.steps import s3_helper
 from frostfs_testlib.steps.cli.container import StorageContainer, StorageContainerInfo, create_container
 from frostfs_testlib.steps.cli.object import get_object, put_object_to_random_node
 from frostfs_testlib.steps.node_management import (
@@ -19,8 +19,6 @@ from frostfs_testlib.steps.node_management import (
     remove_nodes_from_map_morph,
     wait_for_node_to_be_ready,
 )
-from frostfs_testlib.steps.s3 import s3_helper
-from frostfs_testlib.steps.s3.s3_helper import search_nodes_with_bucket
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode, S3Gate, StorageNode
 from frostfs_testlib.storage.controllers import ClusterStateController, ShardsWatcher
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
@@ -137,7 +135,7 @@ class TestFailoverStorage(ClusterTestBase):
             put_object = s3_client.put_object(bucket, file_path)
             s3_helper.check_objects_in_bucket(s3_client, bucket, expected_objects=[file_name])
 
-        node_bucket = search_nodes_with_bucket(
+        node_bucket = s3_helper.search_nodes_with_bucket(
             cluster=self.cluster,
             bucket_name=bucket,
             wallet=default_wallet,

pytest_tests/testsuites/failovers/test_frostfs_failover_network.py

@@ -193,7 +193,7 @@ class TestFailoverNetwork(ClusterTestBase):
             self.tick_epochs(1, alive_node=nodes_without_an_object[0].storage_node, wait_block=2)
 
         with reporter.step("Get object for target nodes to data interfaces, expect false"):
-            with pytest.raises(RuntimeError, match="can't create API client: can't init SDK client: context deadline exceeded"):
+            with pytest.raises(RuntimeError, match="can't create API client: can't init SDK client: context (deadline exceeded|canceled)"):
                 get_object(
                     wallet=default_wallet,
                     cid=storage_object.cid,

pytest_tests/testsuites/metrics/test_container_metrics.py

@@ -14,10 +14,11 @@ from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 from frostfs_testlib.testing.parallel import parallel
 from frostfs_testlib.utils.file_utils import TestFile, generate_file
 
-from ...helpers.container_request import PUBLIC_WITH_POLICY, ContainerRequest, requires_container
+from ...helpers.container_request import PUBLIC_WITH_POLICY, REP_2_1_4_PUBLIC, ContainerRequest, requires_container
 from ...helpers.utility import are_numbers_similar
 
 
+@pytest.mark.order(-5)
 @pytest.mark.nightly
 @pytest.mark.metrics
 class TestContainerMetrics(ClusterTestBase):
@@ -133,7 +134,7 @@ class TestContainerMetrics(ClusterTestBase):
 
     @allure.title("Container size metrics put {objects_count} objects (obj_size={object_size})")
     @pytest.mark.parametrize("objects_count", [5, 10, 20])
-    @requires_container
+    @requires_container(REP_2_1_4_PUBLIC)
     def test_container_size_metrics_more_objects(
         self, object_size: ObjectSize, default_wallet: WalletInfo, objects_count: int, container: str
     ):
@@ -161,13 +162,22 @@ class TestContainerMetrics(ClusterTestBase):
                 tombstones_size += int(tombstone["header"]["payloadLength"])
 
         with reporter.step(f"Check container size metrics, 'should be positive in all nodes'"):
-            futures = parallel(
-                get_metrics_value, self.cluster.cluster_nodes, command="frostfs_node_engine_container_size_bytes", cid=container
-            )
+            with reporter.step("Search container nodes"):
+                container_nodes = search_nodes_with_container(
+                    wallet=default_wallet,
+                    cid=container,
+                    shell=self.shell,
+                    endpoint=self.cluster.default_rpc_endpoint,
+                    cluster=self.cluster,
+                )
+            with reporter.step(f"Get metrics value from container nodes"):
+                futures = parallel(get_metrics_value, container_nodes, command="frostfs_node_engine_container_size_bytes", cid=container)
             metrics_value_nodes = [future.result() for future in futures]
             for act_metric in metrics_value_nodes:
                 assert act_metric >= 0, "Metrics value is negative"
-            assert sum(metrics_value_nodes) // len(self.cluster.cluster_nodes) == tombstones_size, "tomstone size of objects not correct"
+
+        with reporter.step(f"Check container size metrics for tombstone"):
+            assert sum(metrics_value_nodes) // len(container_nodes) == tombstones_size, "tomstone size of objects not correct"
 
     @allure.title("Container metrics (policy={container_request})")
     @pytest.mark.parametrize(

pytest_tests/testsuites/metrics/test_garbage_collector_metrics.py

@@ -16,6 +16,7 @@ from frostfs_testlib.utils.file_utils import generate_file
 from ...helpers.container_request import PUBLIC_WITH_POLICY, requires_container
 
 
+@pytest.mark.order(-9)
 @pytest.mark.nightly
 @pytest.mark.metrics
 class TestGarbageCollectorMetrics(ClusterTestBase):

pytest_tests/testsuites/metrics/test_grpc_metrics.py

@@ -18,6 +18,7 @@ from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 from frostfs_testlib.utils.file_utils import generate_file
 
 
+@pytest.mark.order(-6)
 @pytest.mark.nightly
 @pytest.mark.metrics
 class TestGRPCMetrics(ClusterTestBase):

pytest_tests/testsuites/metrics/test_logs_metrics.py

@@ -14,6 +14,7 @@ from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 from frostfs_testlib.testing.test_control import wait_for_success
 
 
+@pytest.mark.order(-10)
 @pytest.mark.nightly
 @pytest.mark.metrics
 class TestLogsMetrics(ClusterTestBase):
@@ -34,7 +35,9 @@ class TestLogsMetrics(ClusterTestBase):
         current_time = datetime.now(timezone.utc)
         counter_metrics = get_metrics_value(cluster_node, **metrics_greps)
         counter_logs = self.get_count_logs_by_level(cluster_node, metrics_greps.get("level"), restart_time, current_time, log_priority)
-        assert counter_logs == counter_metrics, f"counter_logs: {counter_logs}, counter_metrics: {counter_metrics} in node: {cluster_node}"
+        assert counter_logs == pytest.approx(
+            counter_metrics, rel=0.02
+        ), f"counter_logs: {counter_logs}, counter_metrics: {counter_metrics} in node: {cluster_node}"
 
     @staticmethod
     def get_count_logs_by_level(cluster_node: ClusterNode, log_level: str, after_time: datetime, until_time: datetime, log_priority: str):

pytest_tests/testsuites/metrics/test_object_metrics.py

@@ -17,6 +17,7 @@ from frostfs_testlib.utils.file_utils import TestFile
 from ...helpers.container_request import PUBLIC_WITH_POLICY, ContainerRequest, requires_container
 
 
+@pytest.mark.order(-7)
 @pytest.mark.nightly
 @pytest.mark.metrics
 class TestObjectMetrics(ClusterTestBase):

pytest_tests/testsuites/metrics/test_shard_metrics.py

@@ -19,6 +19,7 @@ from frostfs_testlib.utils.file_utils import generate_file
 from ...helpers.container_request import PUBLIC_WITH_POLICY, requires_container
 
 
+@pytest.mark.order(-8)
 @pytest.mark.nightly
 @pytest.mark.metrics
 class TestShardMetrics(ClusterTestBase):

pytest_tests/testsuites/object/test_object_api_bearer.py

@@ -2,6 +2,7 @@ import allure
 import pytest
 from frostfs_testlib import reporter
 from frostfs_testlib.cli import FrostfsCli
+from frostfs_testlib.resources.cli import FROSTFS_CLI_EXEC
 from frostfs_testlib.shell import Shell
 from frostfs_testlib.steps.cli.container import (
     REP_2_FOR_3_NODES_PLACEMENT_RULE,
@@ -15,6 +16,8 @@ from frostfs_testlib.storage.cluster import Cluster
 from frostfs_testlib.storage.dataclasses import ape
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
+from frostfs_testlib.storage.grpc_operations.client_wrappers import CliClientWrapper
+from frostfs_testlib.storage.grpc_operations.interfaces import GrpcClientWrapper
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 from frostfs_testlib.testing.test_control import expect_not_raises
 from pytest import FixtureRequest
@@ -46,6 +49,11 @@ def bearer_token(frostfs_cli: FrostfsCli, temp_directory: str, user_container: S
     return create_bearer_token(frostfs_cli, temp_directory, user_container.get_id(), rule, cluster.default_rpc_endpoint)
 
 
+@pytest.fixture(scope="session")
+def grpc_client_with_other_wallet(client_shell: Shell, other_wallet: WalletInfo) -> GrpcClientWrapper:
+    return CliClientWrapper(FrostfsCli(client_shell, FROSTFS_CLI_EXEC, other_wallet.config_path))
+
+
 @pytest.fixture()
 def storage_objects(
     user_container: StorageContainer,
@@ -126,6 +134,8 @@ class TestObjectApiWithBearerToken(ClusterTestBase):
                         bearer_token,
                     )
 
+    # TODO: Without PATCH operation,
+    # since it requires specific permissions that do not apply when testing all operations at once
     @allure.title("Wildcard APE rule contains all permissions (obj_size={object_size})")
     def test_ape_wildcard_contains_all_rules(
         self,
@@ -134,5 +144,27 @@ class TestObjectApiWithBearerToken(ClusterTestBase):
         bearer_token: str,
     ):
         obj = storage_objects.pop()
-        with reporter.step(f"Assert all operations available with object"):
+        with reporter.step("Assert all operations available with object"):
             assert_full_access_to_container(other_wallet, obj.cid, obj.oid, obj.file_path, self.shell, self.cluster, bearer_token)
+
+    # ^
+    @allure.title("Wildcard APE rule contains PATCH permission (obj_size={object_size})")
+    def test_ape_wildcard_contains_patch_rule(
+        self,
+        grpc_client_with_other_wallet: GrpcClientWrapper,
+        storage_objects: list[StorageObjectInfo],
+        bearer_token: str,
+    ):
+        obj = storage_objects.pop()
+        with reporter.step("Verify patch is available"):
+            patched_oid = grpc_client_with_other_wallet.object.patch(
+                obj.cid,
+                obj.oid,
+                self.cluster.default_rpc_endpoint,
+                ranges=["99:88"],
+                payloads=[obj.file_path],
+                new_attrs="test-attribute=100",
+                bearer=bearer_token,
+                timeout="200s",
+            )
+            assert patched_oid != obj.oid, "OID of patched object must be different from original one"

pytest_tests/testsuites/object/test_object_without_user.py

@@ -1,5 +1,6 @@
 import logging
 import re
+from typing import Literal
 
 import allure
 import pytest
@@ -8,6 +9,7 @@ from frostfs_testlib.cli import FrostfsCli
 from frostfs_testlib.resources.cli import CLI_DEFAULT_TIMEOUT, FROSTFS_CLI_EXEC
 from frostfs_testlib.resources.error_patterns import OBJECT_IS_LOCKED
 from frostfs_testlib.shell import Shell
+from frostfs_testlib.shell.interfaces import CommandResult
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 from frostfs_testlib.testing.test_control import expect_not_raises
@@ -16,20 +18,24 @@ from frostfs_testlib.utils.file_utils import TestFile, get_file_hash
 logger = logging.getLogger("NeoLogger")
 
 
+def parse_oid(response: CommandResult, response_type: Literal["tombstone", "patch"] = None) -> str:
+    if response_type == "tombstone":
+        id_str = response.stdout.split("\n")[1]
+        oid = id_str.split(":")[1]
+        return oid.strip()
+
+    if response_type == "patch":
+        return response.stdout.split(":")[1].strip()
+
+    id_str = response.stdout.strip().split("\n")[-2]
+    oid = id_str.split(":")[1]
+    return oid.strip()
+
+
 @pytest.mark.nightly
 @pytest.mark.grpc_api
 @pytest.mark.grpc_without_user
 class TestObjectApiWithoutUser(ClusterTestBase):
-    def _parse_oid(self, stdout: str) -> str:
-        id_str = stdout.strip().split("\n")[-2]
-        oid = id_str.split(":")[1]
-        return oid.strip()
-
-    def _parse_tombstone_oid(self, stdout: str) -> str:
-        id_str = stdout.split("\n")[1]
-        tombstone = id_str.split(":")[1]
-        return tombstone.strip()
-
     @pytest.fixture(scope="class")
     def cli_without_wallet(self, client_shell: Shell) -> FrostfsCli:
         return FrostfsCli(client_shell, FROSTFS_CLI_EXEC)
@@ -86,7 +92,7 @@ class TestObjectApiWithoutUser(ClusterTestBase):
                 cli_without_wallet.container.search_node(rpc_endpoint, container, generate_key=True, timeout=CLI_DEFAULT_TIMEOUT)
 
     @allure.title("Put object into public container by native API with generate private key (obj_size={object_size})")
-    def test_put_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, file_path: TestFile, rpc_endpoint: str):
+    def test_put_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, test_file: TestFile, rpc_endpoint: str):
         """
         Validate `object put` into container with public ACL and flag `--generate-key`.
         """
@@ -96,13 +102,13 @@ class TestObjectApiWithoutUser(ClusterTestBase):
                 result = cli_without_wallet.object.put(
                     rpc_endpoint,
                     container,
-                    file_path,
+                    test_file,
                     generate_key=True,
                     no_progress=True,
                     timeout=CLI_DEFAULT_TIMEOUT,
                 )
 
-        oid = self._parse_oid(result.stdout)
+        oid = parse_oid(result)
 
         with reporter.step("List objects with generate key"):
             result = cli_without_wallet.container.list_objects(rpc_endpoint, container, generate_key=True, timeout=CLI_DEFAULT_TIMEOUT)
@@ -112,24 +118,24 @@ class TestObjectApiWithoutUser(ClusterTestBase):
             assert oid in objects, objects
 
     @allure.title("Get public container object by native API with generate private key (obj_size={object_size})")
-    def test_get_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, file_path: TestFile, rpc_endpoint: str):
+    def test_get_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, test_file: TestFile, rpc_endpoint: str):
         """
         Validate `object get` for container with public ACL and flag `--generate-key`.
         """
 
-        expected_hash = get_file_hash(file_path)
+        expected_hash = get_file_hash(test_file)
 
         with reporter.step("Put object with generate key"):
             result = cli_without_wallet.object.put(
                 rpc_endpoint,
                 container,
-                file_path,
+                test_file,
                 generate_key=True,
                 no_progress=True,
                 timeout=CLI_DEFAULT_TIMEOUT,
             )
 
-        oid = self._parse_oid(result.stdout)
+        oid = parse_oid(result)
 
         with reporter.step("Get object with generate key"):
             with expect_not_raises():
@@ -137,19 +143,19 @@ class TestObjectApiWithoutUser(ClusterTestBase):
                     rpc_endpoint,
                     container,
                     oid,
-                    file=file_path,
+                    file=test_file,
                     generate_key=True,
                     no_progress=True,
                     timeout=CLI_DEFAULT_TIMEOUT,
                 )
 
-        downloaded_hash = get_file_hash(file_path)
+        downloaded_hash = get_file_hash(test_file)
 
         with reporter.step("Validate downloaded file"):
             assert expected_hash == downloaded_hash
 
     @allure.title("Head public container object by native API with generate private key (obj_size={object_size})")
-    def test_head_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, file_path: TestFile, rpc_endpoint: str):
+    def test_head_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, test_file: TestFile, rpc_endpoint: str):
         """
         Validate `object head` for container with public ACL and flag `--generate-key`.
         """
@@ -158,20 +164,20 @@ class TestObjectApiWithoutUser(ClusterTestBase):
             result = cli_without_wallet.object.put(
                 rpc_endpoint,
                 container,
-                file_path,
+                test_file,
                 generate_key=True,
                 no_progress=True,
                 timeout=CLI_DEFAULT_TIMEOUT,
             )
 
-        oid = self._parse_oid(result.stdout)
+        oid = parse_oid(result)
 
         with reporter.step("Head object with generate key"):
             with expect_not_raises():
                 cli_without_wallet.object.head(rpc_endpoint, container, oid, generate_key=True, timeout=CLI_DEFAULT_TIMEOUT)
 
     @allure.title("Delete public container object by native API with generate private key (obj_size={object_size})")
-    def test_delete_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, file_path: TestFile, rpc_endpoint: str):
+    def test_delete_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, test_file: TestFile, rpc_endpoint: str):
         """
         Validate `object delete` for container with public ACL and flag `--generate key`.
         """
@@ -180,19 +186,19 @@ class TestObjectApiWithoutUser(ClusterTestBase):
             result = cli_without_wallet.object.put(
                 rpc_endpoint,
                 container,
-                file_path,
+                test_file,
                 generate_key=True,
                 no_progress=True,
                 timeout=CLI_DEFAULT_TIMEOUT,
             )
 
-        oid = self._parse_oid(result.stdout)
+        oid = parse_oid(result)
 
         with reporter.step("Delete object with generate key"):
             with expect_not_raises():
                 result = cli_without_wallet.object.delete(rpc_endpoint, container, oid, generate_key=True, timeout=CLI_DEFAULT_TIMEOUT)
 
-        oid = self._parse_tombstone_oid(result.stdout)
+        oid = parse_oid(result, response_type="tombstone")
 
         with reporter.step("Head object with generate key"):
             result = cli_without_wallet.object.head(
@@ -207,8 +213,37 @@ class TestObjectApiWithoutUser(ClusterTestBase):
             object_type = re.search(r"(?<=type: )tombstone", result.stdout, re.IGNORECASE).group()
             assert object_type == "TOMBSTONE", object_type
 
+    @allure.title("Patch object in public container with generate private key (obj_size={object_size})")
+    def test_patch_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, test_file: TestFile, rpc_endpoint: str):
+        with reporter.step("Put object with generate key"):
+            result = cli_without_wallet.object.put(
+                rpc_endpoint,
+                container,
+                test_file,
+                generate_key=True,
+                no_progress=True,
+                timeout=CLI_DEFAULT_TIMEOUT,
+            )
+
+        oid = parse_oid(result)
+
+        with reporter.step("Patch object with generate key"):
+            with expect_not_raises():
+                result = cli_without_wallet.object.patch(
+                    rpc_endpoint,
+                    container,
+                    oid,
+                    ["0:500"],
+                    [test_file],
+                    generate_key=True,
+                    timeout=CLI_DEFAULT_TIMEOUT,
+                )
+
+            patched_oid = parse_oid(result, response_type="patch")
+            assert oid != patched_oid, "Patched object must have new object id"
+
     @allure.title("Lock public container object by native API with generate private key (obj_size={object_size})")
-    def test_lock_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, file_path: TestFile, rpc_endpoint: str):
+    def test_lock_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, test_file: TestFile, rpc_endpoint: str):
         """
         Validate `object lock` for container with public ACL and flag `--generate-key`.
         Attempt to delete the locked object.
@@ -218,13 +253,13 @@ class TestObjectApiWithoutUser(ClusterTestBase):
             result = cli_without_wallet.object.put(
                 rpc_endpoint,
                 container,
-                file_path,
+                test_file,
                 generate_key=True,
                 no_progress=True,
                 timeout=CLI_DEFAULT_TIMEOUT,
             )
 
-        oid = self._parse_oid(result.stdout)
+        oid = parse_oid(result)
 
         with reporter.step("Lock object with generate key"):
             with expect_not_raises():
@@ -248,7 +283,7 @@ class TestObjectApiWithoutUser(ClusterTestBase):
                 )
 
     @allure.title("Search public container objects by native API with generate private key (obj_size={object_size})")
-    def test_search_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, file_path: TestFile, rpc_endpoint: str):
+    def test_search_object_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, test_file: TestFile, rpc_endpoint: str):
         """
         Validate `object search` for container with public ACL and flag `--generate-key`.
         """
@@ -257,13 +292,13 @@ class TestObjectApiWithoutUser(ClusterTestBase):
             result = cli_without_wallet.object.put(
                 rpc_endpoint,
                 container,
-                file_path,
+                test_file,
                 generate_key=True,
                 no_progress=True,
                 timeout=CLI_DEFAULT_TIMEOUT,
             )
 
-        oid = self._parse_oid(result.stdout)
+        oid = parse_oid(result)
 
         with reporter.step("Object search with generate key"):
             with expect_not_raises():
@@ -274,7 +309,7 @@ class TestObjectApiWithoutUser(ClusterTestBase):
             assert oid in object_ids
 
     @allure.title("Get range of public container object by native API with generate private key (obj_size={object_size})")
-    def test_range_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, file_path: TestFile, rpc_endpoint: str):
+    def test_range_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, test_file: TestFile, rpc_endpoint: str):
         """
         Validate `object range` for container with public ACL and `--generate-key`.
         """
@@ -283,13 +318,13 @@ class TestObjectApiWithoutUser(ClusterTestBase):
             result = cli_without_wallet.object.put(
                 rpc_endpoint,
                 container,
-                file_path,
+                test_file,
                 generate_key=True,
                 no_progress=True,
                 timeout=CLI_DEFAULT_TIMEOUT,
             )
 
-        oid = self._parse_oid(result.stdout)
+        oid = parse_oid(result)
 
         with reporter.step("Get range of object with generate key"):
             with expect_not_raises():
@@ -298,13 +333,13 @@ class TestObjectApiWithoutUser(ClusterTestBase):
                     container,
                     oid,
                     "0:10",
-                    file=file_path,
+                    file=test_file,
                     generate_key=True,
                     timeout=CLI_DEFAULT_TIMEOUT,
                 )
 
     @allure.title("Get hash of public container object by native API with generate private key (obj_size={object_size})")
-    def test_hash_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, file_path: TestFile, rpc_endpoint: str):
+    def test_hash_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, test_file: TestFile, rpc_endpoint: str):
         """
         Validate `object hash` for container with public ACL and `--generate-key`.
         """
@@ -313,13 +348,13 @@ class TestObjectApiWithoutUser(ClusterTestBase):
             result = cli_without_wallet.object.put(
                 rpc_endpoint,
                 container,
-                file_path,
+                test_file,
                 generate_key=True,
                 no_progress=True,
                 timeout=CLI_DEFAULT_TIMEOUT,
             )
 
-        oid = self._parse_oid(result.stdout)
+        oid = parse_oid(result)
 
         with reporter.step("Get range hash of object with generate key"):
             with expect_not_raises():
@@ -333,7 +368,7 @@ class TestObjectApiWithoutUser(ClusterTestBase):
                 )
 
     @allure.title("Get public container object nodes by native API with generate private key (obj_size={object_size})")
-    def test_nodes_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, file_path: TestFile, rpc_endpoint: str):
+    def test_nodes_with_generate_key(self, cli_without_wallet: FrostfsCli, container: str, test_file: TestFile, rpc_endpoint: str):
         """
         Validate `object nodes` for container with public ACL and `--generate-key`.
         """
@@ -342,13 +377,13 @@ class TestObjectApiWithoutUser(ClusterTestBase):
             result = cli_without_wallet.object.put(
                 rpc_endpoint,
                 container,
-                file_path,
+                test_file,
                 no_progress=True,
                 generate_key=True,
                 timeout=CLI_DEFAULT_TIMEOUT,
             )
 
-        oid = self._parse_oid(result.stdout)
+        oid = parse_oid(result)
 
         with reporter.step("Configure frostfs-cli for alive remote node"):
             alive_node = self.cluster.cluster_nodes[0]

pytest_tests/testsuites/replication/test_ec_replication.py

@@ -7,10 +7,10 @@ import yaml
 from frostfs_testlib import reporter
 from frostfs_testlib.cli import FrostfsAdm, FrostfsCli
 from frostfs_testlib.cli.netmap_parser import NetmapParser
+from frostfs_testlib.clients import AwsCliClient, S3ClientWrapper
+from frostfs_testlib.clients.s3 import BucketContainerResolver, VersioningStatus
 from frostfs_testlib.resources.cli import CLI_DEFAULT_TIMEOUT, FROSTFS_ADM_CONFIG_PATH, FROSTFS_ADM_EXEC, FROSTFS_CLI_EXEC
 from frostfs_testlib.resources.common import COMPLEX_OBJECT_CHUNKS_COUNT, COMPLEX_OBJECT_TAIL_SIZE, HOSTING_CONFIG_FILE, MORPH_BLOCK_TIME
-from frostfs_testlib.s3 import AwsCliClient, S3ClientWrapper
-from frostfs_testlib.s3.interfaces import BucketContainerResolver, VersioningStatus
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode, StorageNode
 from frostfs_testlib.storage.controllers import ClusterStateController
 from frostfs_testlib.storage.controllers.state_managers.config_state_manager import ConfigStateManager

pytest_tests/testsuites/services/http_gate/test_http_bearer.py

@@ -5,7 +5,7 @@ import pytest
 from frostfs_testlib import reporter
 from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
 from frostfs_testlib.steps.acl import bearer_token_base64_from_file
-from frostfs_testlib.steps.http.http_gate import upload_via_http_gate_curl, verify_object_hash
+from frostfs_testlib.steps.http_gate import upload_via_http_gate_curl, verify_object_hash
 from frostfs_testlib.storage.cluster import Cluster
 from frostfs_testlib.storage.dataclasses import ape
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize

pytest_tests/testsuites/services/http_gate/test_http_gate.py

@@ -3,7 +3,7 @@ import pytest
 from frostfs_testlib import reporter
 from frostfs_testlib.steps.cli.object import put_object_to_random_node
 from frostfs_testlib.steps.epoch import get_epoch
-from frostfs_testlib.steps.http.http_gate import (
+from frostfs_testlib.steps.http_gate import (
     attr_into_header,
     get_object_by_attr_and_verify_hashes,
     get_via_http_curl,

pytest_tests/testsuites/services/http_gate/test_http_headers.py

@@ -6,7 +6,7 @@ import pytest
 from frostfs_testlib import reporter
 from frostfs_testlib.steps.cli.container import delete_container, list_containers, wait_for_container_deletion
 from frostfs_testlib.steps.cli.object import delete_object
-from frostfs_testlib.steps.http.http_gate import (
+from frostfs_testlib.steps.http_gate import (
     attr_into_str_header_curl,
     get_object_by_attr_and_verify_hashes,
     try_to_get_object_and_expect_error,

pytest_tests/testsuites/services/http_gate/test_http_object.py

@@ -3,24 +3,26 @@ import logging
 import allure
 import pytest
 from frostfs_testlib import reporter
-from frostfs_testlib.s3 import AwsCliClient, S3ClientWrapper
+from frostfs_testlib.clients import S3ClientWrapper
+from frostfs_testlib.steps import s3_helper
 from frostfs_testlib.steps.cli.object import put_object_to_random_node
-from frostfs_testlib.steps.http.http_gate import (
+from frostfs_testlib.steps.http_gate import (
     assert_hashes_are_equal,
     get_object_by_attr_and_verify_hashes,
     get_via_http_gate,
     try_to_get_object_via_passed_request_and_expect_error,
     verify_object_hash,
 )
-from frostfs_testlib.steps.s3 import s3_helper
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
-from frostfs_testlib.utils.file_utils import TestFile
+from frostfs_testlib.utils.file_utils import TestFile, generate_file, split_file
 
 from ....helpers.container_request import REP_2_1_4_PUBLIC, requires_container
 
 logger = logging.getLogger("NeoLogger")
 
+PART_SIZE = 5 * 1024 * 1024
+
 
 @pytest.mark.nightly
 @pytest.mark.sanity
@@ -66,6 +68,7 @@ class Test_http_object(ClusterTestBase):
                 cluster=self.cluster,
                 attributes=f"{key_value1},{key_value2}",
             )
+
         with reporter.step("Get object and verify hashes [ get/$CID/$OID ]"):
             verify_object_hash(
                 oid=oid,
@@ -91,25 +94,19 @@ class Test_http_object(ClusterTestBase):
             )
 
         with reporter.step("Download the object with attribute [get_by_attribute/$CID/chapter1/peace]"):
-            get_object_by_attr_and_verify_hashes(
-                oid=oid,
-                file_name=test_file.path,
-                cid=container,
-                attrs=attrs,
-                node=self.cluster.cluster_nodes[0],
-            )
+            get_object_by_attr_and_verify_hashes(oid, test_file, container, attrs, self.cluster.cluster_nodes[0])
+
         with reporter.step("[Negative] try to get object: get_by_attribute/$CID/$OID"):
             request = f"/get_by_attribute/{container}/{oid}"
             try_to_get_object_via_passed_request_and_expect_error(
-                cid=container,
-                oid=oid,
-                node=self.cluster.cluster_nodes[0],
+                container,
+                oid,
+                self.cluster.cluster_nodes[0],
                 error_pattern=expected_err_msg,
                 http_request_path=request,
             )
 
-    @allure.title("Put over s3, Get over HTTP with bucket name and key (object_size={object_size})")
-    @pytest.mark.parametrize("s3_client", [AwsCliClient], indirect=True)
+    @allure.title("Put object over S3, get over HTTP with bucket name and key (s3_client={s3_client}, object_size={object_size})")
     def test_object_put_get_bucketname_key(self, test_file: TestFile, s3_client: S3ClientWrapper):
         """
         Test that object can be put using s3-gateway interface and got via HTTP with bucket name and object key.
@@ -125,17 +122,55 @@ class Test_http_object(ClusterTestBase):
         Hashes must be the same.
         """
 
-        object_key = s3_helper.object_key_from_file_path(test_file.path)
-        bucket = s3_client.create_bucket(acl="public-read-write")
-        s3_client.put_object(bucket=bucket, filepath=test_file.path, key=object_key)
-        obj_s3 = s3_client.get_object(bucket=bucket, key=object_key)
+        object_key = s3_helper.object_key_from_file_path(test_file)
 
-        request = f"/get/{bucket}/{object_key}"
-        obj_http = get_via_http_gate(
-            cid=None,
-            oid=None,
-            node=self.cluster.cluster_nodes[0],
-            request_path=request,
-        )
-        with reporter.step("Verify hashes"):
+        with reporter.step("Create public bucket"):
+            bucket = s3_client.create_bucket(acl="public-read-write")
+
+        with reporter.step("Put object"):
+            s3_client.put_object(bucket, test_file, object_key)
+
+        with reporter.step("Get object via S3 gate"):
+            obj_s3 = s3_client.get_object(bucket, object_key)
+
+        with reporter.step("Get object via HTTP gate"):
+            obj_http = get_via_http_gate(bucket, object_key, node=self.cluster.cluster_nodes[0])
+
+        with reporter.step("Make sure the hashes of both objects are the same"):
             assert_hashes_are_equal(test_file.path, obj_http, obj_s3)
+
+    @allure.title("Put multipart object over S3, get over HTTP with bucket name and key (s3_client={s3_client})")
+    def test_object_put_get_bucketname_key_multipart(self, s3_client: S3ClientWrapper):
+        parts = []
+        parts_count = 5
+        original_size = PART_SIZE * parts_count
+
+        with reporter.step("Create public container"):
+            bucket = s3_client.create_bucket(acl="public-read-write")
+
+        with reporter.step("Generate original object and split it into parts"):
+            original_file = generate_file(original_size)
+            file_parts = split_file(original_file, parts_count)
+            object_key = s3_helper.object_key_from_file_path(original_file)
+
+        with reporter.step("Create multipart and upload parts"):
+            upload_id = s3_client.create_multipart_upload(bucket, object_key)
+            for part_id, file_path in enumerate(file_parts, start=1):
+                etag = s3_client.upload_part(bucket, object_key, upload_id, part_id, file_path)
+                parts.append((part_id, etag))
+
+        with reporter.step("Check all parts are visible in bucket"):
+            got_parts = s3_client.list_parts(bucket, object_key, upload_id)
+            assert len(got_parts) == len(file_parts), f"Expected {parts_count} parts, got:\n{got_parts}"
+
+        with reporter.step("Complete multipart upload"):
+            s3_client.complete_multipart_upload(bucket, object_key, upload_id, parts)
+
+        with reporter.step("Get multipart object via S3 gate"):
+            obj_s3 = s3_client.get_object(bucket, object_key)
+
+        with reporter.step("Get multipart object via HTTP gate"):
+            obj_http = get_via_http_gate(bucket, object_key, self.cluster.cluster_nodes[0])
+
+        with reporter.step("Make sure the hashes of both objects are the same"):
+            assert_hashes_are_equal(original_file, obj_http, obj_s3)

pytest_tests/testsuites/services/http_gate/test_http_streaming.py

@@ -3,7 +3,7 @@ import logging
 import allure
 import pytest
 from frostfs_testlib import reporter
-from frostfs_testlib.steps.http.http_gate import upload_via_http_gate_curl, verify_object_hash
+from frostfs_testlib.steps.http_gate import upload_via_http_gate_curl, verify_object_hash
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase

pytest_tests/testsuites/services/http_gate/test_http_system_header.py

@@ -9,7 +9,7 @@ from frostfs_testlib import reporter
 from frostfs_testlib.resources.error_patterns import OBJECT_NOT_FOUND
 from frostfs_testlib.steps.cli.object import get_netmap_netinfo, get_object_from_random_node, head_object
 from frostfs_testlib.steps.epoch import get_epoch, wait_for_epochs_align
-from frostfs_testlib.steps.http.http_gate import (
+from frostfs_testlib.steps.http_gate import (
     attr_into_str_header_curl,
     try_to_get_object_and_expect_error,
     upload_via_http_gate_curl,

pytest_tests/testsuites/services/s3_gate/test_s3_ACL.py

@@ -1,10 +1,10 @@
 import allure
 import pytest
 from frostfs_testlib import reporter
+from frostfs_testlib.clients import S3ClientWrapper
 from frostfs_testlib.resources.error_patterns import S3_BUCKET_DOES_NOT_ALLOW_ACL
 from frostfs_testlib.resources.s3_acl_grants import PRIVATE_GRANTS, PUBLIC_READ_GRANTS, PUBLIC_READ_WRITE_GRANTS
-from frostfs_testlib.s3 import S3ClientWrapper
-from frostfs_testlib.steps.s3 import s3_helper
+from frostfs_testlib.steps import s3_helper
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
 from frostfs_testlib.utils.file_utils import generate_file
 
@@ -33,32 +33,32 @@ class TestS3GateACL:
     def test_s3_create_bucket_with_ACL(self, s3_client: S3ClientWrapper):
         with reporter.step("Create bucket with ACL private"):
             bucket = s3_client.create_bucket(object_lock_enabled_for_bucket=True, acl="private")
-            bucket_grants = s3_client.get_bucket_acl(bucket)
+            bucket_grants = s3_client.get_bucket_acl(bucket).get("Grants")
             s3_helper.verify_acl_permissions(bucket_grants, PRIVATE_GRANTS)
 
         with reporter.step("Create bucket with ACL public-read"):
             read_bucket = s3_client.create_bucket(object_lock_enabled_for_bucket=True, acl="public-read")
-            bucket_grants = s3_client.get_bucket_acl(read_bucket)
+            bucket_grants = s3_client.get_bucket_acl(read_bucket).get("Grants")
             s3_helper.verify_acl_permissions(bucket_grants, PUBLIC_READ_GRANTS)
 
         with reporter.step("Create bucket with ACL public-read-write"):
             public_rw_bucket = s3_client.create_bucket(object_lock_enabled_for_bucket=True, acl="public-read-write")
-            bucket_grants = s3_client.get_bucket_acl(public_rw_bucket)
+            bucket_grants = s3_client.get_bucket_acl(public_rw_bucket).get("Grants")
             s3_helper.verify_acl_permissions(bucket_grants, PUBLIC_READ_WRITE_GRANTS)
 
     @allure.title("Bucket ACL (s3_client={s3_client})")
     def test_s3_bucket_ACL(self, s3_client: S3ClientWrapper):
         with reporter.step("Create bucket with public-read-write ACL"):
             bucket = s3_client.create_bucket(object_lock_enabled_for_bucket=True, acl="public-read-write")
-            bucket_grants = s3_client.get_bucket_acl(bucket)
+            bucket_grants = s3_client.get_bucket_acl(bucket).get("Grants")
             s3_helper.verify_acl_permissions(bucket_grants, PUBLIC_READ_WRITE_GRANTS)
 
         with reporter.step("Change bucket ACL to private"):
             s3_client.put_bucket_acl(bucket, acl="private")
-            bucket_grants = s3_client.get_bucket_acl(bucket)
+            bucket_grants = s3_client.get_bucket_acl(bucket).get("Grants")
             s3_helper.verify_acl_permissions(bucket_grants, PRIVATE_GRANTS)
 
         with reporter.step("Change bucket ACL to public-read"):
             s3_client.put_bucket_acl(bucket, acl="public-read")
-            bucket_grants = s3_client.get_bucket_acl(bucket)
+            bucket_grants = s3_client.get_bucket_acl(bucket).get("Grants")
             s3_helper.verify_acl_permissions(bucket_grants, PUBLIC_READ_GRANTS)

pytest_tests/testsuites/services/s3_gate/test_s3_bucket.py

@@ -4,8 +4,8 @@ from datetime import datetime, timedelta
 import allure
 import pytest
 from frostfs_testlib import reporter
-from frostfs_testlib.s3 import S3ClientWrapper, VersioningStatus
-from frostfs_testlib.steps.s3 import s3_helper
+from frostfs_testlib.clients.s3 import S3ClientWrapper, VersioningStatus
+from frostfs_testlib.steps import s3_helper
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
 from frostfs_testlib.utils import string_utils
 from frostfs_testlib.utils.file_utils import generate_file

pytest_tests/testsuites/services/s3_gate/test_s3_http_object.py

@@ -0,0 +1,623 @@
+import random
+import time
+from datetime import datetime
+from email.utils import formatdate
+
+import allure
+import pytest
+from frostfs_testlib import reporter
+from frostfs_testlib.cli.generic_cli import GenericCli
+from frostfs_testlib.clients import Boto3ClientWrapper, S3ClientWrapper, S3HttpClient
+from frostfs_testlib.clients.s3 import VersioningStatus
+from frostfs_testlib.credentials.interfaces import CredentialsProvider, User
+from frostfs_testlib.shell.interfaces import CommandOptions
+from frostfs_testlib.shell.local_shell import LocalShell
+from frostfs_testlib.steps import s3_helper
+from frostfs_testlib.storage.cluster import Cluster, ClusterNode
+from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
+from frostfs_testlib.storage.dataclasses.policy import PlacementPolicy
+from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
+from frostfs_testlib.utils.file_utils import TestFile, generate_file, get_file_hash, split_file
+from frostfs_testlib.utils.string_utils import unique_name
+
+from ....resources.common import S3_POLICY_FILE_LOCATION
+
+FIVE_GIGABYTES = 5_368_709_120
+PART_SIZE_FOR_MULTIPART = 5 * 1024 * 1024
+
+
+@reporter.step("Allow patch for bucket")
+def allow_patch_for_bucket(s3_client: S3ClientWrapper, bucket: str):
+    s3_client.put_bucket_policy(
+        bucket,
+        policy={
+            "Version": "2012-10-17",
+            "Id": "aaaa-bbbb-cccc-dddd",
+            "Statement": [
+                {
+                    "Sid": "AddPerm",
+                    "Effect": "Allow",
+                    "Principal": "*",
+                    "Action": ["s3:PatchObject"],
+                    "Resource": [f"arn:aws:s3:::{bucket}/*"],
+                },
+            ],
+        },
+    )
+
+
+def pytest_generate_tests(metafunc: pytest.Metafunc):
+    if "s3_client" not in metafunc.fixturenames:
+        return
+    metafunc.parametrize("s3_policy", [S3_POLICY_FILE_LOCATION], ids=["s3policy"], indirect=True)
+
+
+@pytest.fixture(scope="session", params=[pytest.param("rep3", marks=pytest.mark.rep), pytest.param("ec3.1", marks=pytest.mark.ec)])
+def placement_policy(request: pytest.FixtureRequest) -> PlacementPolicy:
+    if request.param == "ec3.1":
+        return PlacementPolicy("ec3.1", "ec3.1")
+    return PlacementPolicy("rep3", "rep3")
+
+
+@pytest.fixture(scope="session")
+def versioning_status(request: pytest.FixtureRequest) -> VersioningStatus:
+    if "param" in request.__dict__:
+        return request.param
+    return VersioningStatus.UNDEFINED
+
+
+@allure.title("[Class] Create bucket")
+@pytest.fixture(scope="class")
+def bucket(s3_client: S3ClientWrapper, versioning_status: VersioningStatus, placement_policy: PlacementPolicy) -> str:
+    with reporter.step(f"Create bucket with location constraint {placement_policy.value}"):
+        bucket = s3_client.create_bucket(location_constraint=placement_policy.value)
+
+    s3_helper.set_bucket_versioning(s3_client, bucket, versioning_status)
+    allow_patch_for_bucket(s3_client, bucket)
+    return bucket
+
+
+@pytest.fixture(scope="function")
+def original_object(s3_client: S3ClientWrapper, bucket: str, test_file: TestFile) -> str:
+    with reporter.step("Put object"):
+        key = s3_helper.object_key_from_file_path(test_file)
+        s3_client.put_object(bucket, test_file, key)
+
+    return key
+
+
+@allure.title("[Session]: Create S3 client for another user")
+@pytest.fixture(scope="session")
+def another_s3_client(
+    users_pool: list[User],
+    s3_policy: str | None,
+    cluster: Cluster,
+    credentials_provider: CredentialsProvider,
+    s3_client: S3ClientWrapper,
+) -> S3ClientWrapper:
+    user = users_pool[0]
+    node = cluster.cluster_nodes[0]
+    credentials_provider.S3.provide(user, node, s3_policy)
+    s3_client_cls = type(s3_client)
+    return s3_client_cls(user.s3_credentials.access_key, user.s3_credentials.secret_key, cluster.default_s3_gate_endpoint)
+
+
+@allure.title("[Class] Create bucket under another user")
+@pytest.fixture(scope="class")
+def another_bucket(another_s3_client: S3ClientWrapper, versioning_status: VersioningStatus, placement_policy: PlacementPolicy) -> str:
+    with reporter.step(f"Create bucket with location constraint {placement_policy.value}"):
+        bucket = another_s3_client.create_bucket(location_constraint=placement_policy.value)
+
+    s3_helper.set_bucket_versioning(another_s3_client, bucket, versioning_status)
+    allow_patch_for_bucket(another_s3_client, bucket)
+    return bucket
+
+
+@pytest.mark.nightly
+@pytest.mark.s3_gate
+class TestS3GateHttpObject(ClusterTestBase):
+    @allure.title("Patch simple object payload (range={patch_range}, s3_client={s3_client}, policy={placement_policy})")
+    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
+    @pytest.mark.parametrize(
+        "patch_range",
+        # String "object" denotes size of object.
+        ["0:19", "500:550", "object/2-100:object/2+200", "object-1:object", "object:object", "object:object+123"],
+    )
+    def test_patch_simple_object_payload(
+        self,
+        s3_client: S3ClientWrapper,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        original_object: str,
+        object_size: ObjectSize,
+        patch_range: str,
+    ):
+        start, end = s3_helper.get_range_relative_to_object(patch_range, object_size.value, int_values=True)
+        content_size = end - start + 1
+        content_range = f"bytes {start}-{end}/*"
+
+        with reporter.step("Generate payload object"):
+            content_file = generate_file(content_size)
+
+        with reporter.step("Patch simple object"):
+            s3_http_client.patch_object(bucket, original_object, content_file, content_range)
+
+        with reporter.step("Get patched part of object and make sure it has changed correctly"):
+            patched_file_part = s3_client.get_object(bucket, original_object, object_range=(start, end))
+            assert get_file_hash(patched_file_part) == get_file_hash(
+                content_file
+            ), "Expected content hash did not match actual content hash"
+
+    @allure.title("Patch complex object payload (range={patch_range}, s3_client={s3_client}, policy={placement_policy})")
+    @pytest.mark.parametrize("object_size", ["complex"], indirect=True)
+    @pytest.mark.parametrize(
+        "patch_range",
+        # Strings "object" and "part" denote size of object and its part, respectively.
+        ["part:part+100", "object-part:object", "0:part", "part*2:part*3", "part-1:part*2", "part+1:part*2-1"],
+    )
+    def test_patch_complex_object_payload(
+        self,
+        s3_client: S3ClientWrapper,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        original_object: str,
+        object_size: ObjectSize,
+        max_object_size: int,
+        patch_range: str,
+    ):
+        start, end = s3_helper.get_range_relative_to_object(patch_range, object_size.value, max_object_size, int_values=True)
+        content_size = end - start + 1
+        content_range = f"bytes {start}-{end}/*"
+
+        with reporter.step("Generate payload object"):
+            content_file = generate_file(content_size)
+
+        with reporter.step("Patch complex object"):
+            s3_http_client.patch_object(bucket, original_object, content_file, content_range)
+
+        with reporter.step("Get patched part of object and make sure it has changed correctly"):
+            patched_file_part = s3_client.get_object(bucket, original_object, object_range=(start, end))
+            assert get_file_hash(patched_file_part) == get_file_hash(
+                content_file
+            ), "Expected content hash did not match actual content hash"
+
+    @allure.title(
+        "Patch object with fulfilled If-Match condition (s3_client={s3_client}, object_size={object_size}, policy={placement_policy})"
+    )
+    def test_patch_with_fulfilled_if_match_contidion(
+        self,
+        s3_client: S3ClientWrapper,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        original_object: str,
+    ):
+        start, end = 100, 199
+        content_size = end - start + 1
+        content_range = f"bytes {start}-{end}/*"
+
+        with reporter.step("Generate payload object"):
+            content_file = generate_file(content_size)
+            expected_hash = get_file_hash(content_file)
+
+        with reporter.step("Get object ETag attribute"):
+            object_info = s3_client.head_object(bucket, original_object)
+            etag = object_info["ETag"]
+
+        with reporter.step("Patch object with If-Match header"):
+            s3_http_client.patch_object(bucket, original_object, content_file, content_range, if_match=etag)
+
+        with reporter.step("Get patched object and make sure it has changed correctly"):
+            patched_file = s3_client.get_object(bucket, original_object)
+            patched_hash = get_file_hash(patched_file, offset=start, len=content_size)
+            assert patched_hash == expected_hash, "Expected content hash did not match actual content hash"
+
+    @allure.title(
+        "[NEGATIVE] Patch cannot be applied with failed If-Match condition "
+        "(s3_client={s3_client}, object_size={object_size}, policy={placement_policy})"
+    )
+    def test_patch_with_failed_if_match_condition(self, s3_http_client: S3HttpClient, bucket: str, original_object: str):
+        with reporter.step("Try patch object with If-Match header and get exception"):
+            with pytest.raises(Exception, match="PreconditionFailed"):
+                s3_http_client.patch_object(bucket, original_object, "content", "bytes 0-6/*", if_match="nonexistentetag")
+
+    @allure.title(
+        "Patch object with fulfilled If-Unmodified-Since condition "
+        "(s3_client={s3_client}, object_size={object_size}, policy={placement_policy})"
+    )
+    def test_patch_with_fulfilled_if_unmodified_since_condition(
+        self,
+        s3_client: S3ClientWrapper,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        original_object: str,
+    ):
+        start, end = 235, 341
+        content_size = end - start + 1
+        content_range = f"bytes {start}-{end}/*"
+
+        with reporter.step("Generate payload object"):
+            content_file = generate_file(content_size)
+            expected_hash = get_file_hash(content_file)
+
+        with reporter.step("Get object LastModified attribute"):
+            response = s3_client.head_object(bucket, original_object)
+            if isinstance(response["LastModified"], str):
+                response["LastModified"] = datetime.fromisoformat(response["LastModified"])
+
+            # Convert datetime to RFC 7232 format
+            last_modified = formatdate(response["LastModified"].timestamp(), localtime=False, usegmt=True)
+
+        with reporter.step("Patch object with If-Unmodified-Since header"):
+            s3_http_client.patch_object(bucket, original_object, content_file, content_range, if_unmodified_since=last_modified)
+
+        with reporter.step("Get patched object and make sure it has changed correctly"):
+            patched_file = s3_client.get_object(bucket, original_object)
+            patched_hash = get_file_hash(patched_file, offset=start, len=content_size)
+            assert patched_hash == expected_hash, "Expected content hash did not match actual content hash"
+
+    @allure.title(
+        "[NEGATIVE] Patch cannot be applied with failed If-Unmodified-Since condition "
+        "(s3_client={s3_client}, object_size={object_size}, policy={placement_policy})"
+    )
+    def test_patch_with_failed_if_unmodified_since_condition(
+        self,
+        s3_client: S3ClientWrapper,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        original_object: str,
+        test_file: TestFile,
+    ):
+        with reporter.step("Get original object LastModified attribute"):
+            response = s3_client.head_object(bucket, original_object)
+            if isinstance(response["LastModified"], str):
+                response["LastModified"] = datetime.fromisoformat(response["LastModified"])
+
+            # Convert datetime to RFC 7232 format
+            previous_last_modified = formatdate(response["LastModified"].timestamp(), localtime=False, usegmt=True)
+
+        with reporter.step("Wait two seconds for LastModified to update"):
+            # Next PUT for a simple object occurs at the same second the object was initially loaded,
+            # so the LastModified attribute "as if" does not change after the operation.
+            time.sleep(2)
+
+        with reporter.step("Put new data for existing object"):
+            s3_client.put_object(bucket, test_file, original_object)
+
+        with reporter.step("Get object LastModified attribute with new data and make sure it has changed"):
+            response = s3_client.head_object(bucket, original_object)
+            if isinstance(response["LastModified"], str):
+                response["LastModified"] = datetime.fromisoformat(response["LastModified"])
+
+            # Convert datetime to RFC 7232 format
+            last_modified = formatdate(response["LastModified"].timestamp(), localtime=False, usegmt=True)
+            assert last_modified != previous_last_modified, f"Attribute LastModified was expected to change: {last_modified}"
+
+        with reporter.step("Try patch object with If-Unmodified-Since header and get exception"):
+            with pytest.raises(Exception, match="PreconditionFailed"):
+                s3_http_client.patch_object(bucket, original_object, b"modify", "bytes 0-5/*", if_unmodified_since=previous_last_modified)
+
+    @allure.title(
+        "Patch object with fulfilled x-amz-expected-bucket-owner condition "
+        "(s3_client={s3_client}, object_size={object_size}, policy={placement_policy})"
+    )
+    def test_patch_with_fulfilled_if_expected_bucket_owner_condition(
+        self,
+        s3_client: S3ClientWrapper,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        original_object: str,
+    ):
+        start, end = 512, 749
+        content_size = end - start + 1
+        content_range = f"bytes {start}-{end}/*"
+
+        with reporter.step("Generate payload object"):
+            content_file = generate_file(content_size)
+            expected_hash = get_file_hash(content_file)
+
+        with reporter.step("Get bucket owner ID"):
+            bucket_acl = s3_client.get_bucket_acl(bucket)
+            expected_bucket_owner = bucket_acl["Owner"]["DisplayName"]
+
+        with reporter.step("Patch object with x-amz-expected-bucket-owner header"):
+            s3_http_client.patch_object(
+                bucket,
+                original_object,
+                content_file,
+                content_range,
+                x_amz_expected_bucket_owner=expected_bucket_owner,
+            )
+
+        with reporter.step("Get patched object and make sure it has changed correctly"):
+            patched_file = s3_client.get_object(bucket, original_object)
+            patched_hash = get_file_hash(patched_file, offset=start, len=content_size)
+            assert patched_hash == expected_hash, "Expected content hash did not match actual content hash"
+
+    @allure.title(
+        "[NEGATIVE] Patch cannot be applied with non-existent bucket owner ID in x-amz-expected-bucket-owner header "
+        "(s3_client={s3_client}, object_size={object_size}, policy={placement_policy})"
+    )
+    def test_patch_with_non_existent_bucket_owner_id(
+        self,
+        s3_client: S3ClientWrapper,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        original_object: str,
+    ):
+        with reporter.step("Get bucket owner ID"):
+            bucket_acl = s3_client.get_bucket_acl(bucket)
+            bucket_owner = bucket_acl["Owner"]["DisplayName"]
+
+        with reporter.step("Change owner ID to non-existent"):
+            unexpected_bucket_owner = list(bucket_owner)
+            random.shuffle(unexpected_bucket_owner)
+            unexpected_bucket_owner = "".join(unexpected_bucket_owner)
+
+        with reporter.step("Try patch object with x-amz-expected-bucket-owner header and get exception"):
+            with pytest.raises(Exception, match="AccessDenied"):
+                s3_http_client.patch_object(
+                    bucket,
+                    original_object,
+                    b"blablabla",
+                    "bytes 10-18/*",
+                    x_amz_expected_bucket_owner=unexpected_bucket_owner,
+                )
+
+    # AwsCliClient is not configured correctly for some cases,
+    # resulting in both buckets being created from the same user, which causes an error.
+    @allure.title(
+        "[NEGATIVE] Patch cannot be applied with another bucket owner ID in x-amz-expected-bucket-owner header "
+        "(s3_client={s3_client}, object_size={object_size}, policy={placement_policy})"
+    )
+    @pytest.mark.parametrize("s3_client", [Boto3ClientWrapper], indirect=True)
+    def test_patch_with_another_bucket_owner_id(
+        self,
+        s3_client: S3ClientWrapper,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        original_object: str,
+        another_bucket: str,
+    ):
+        with reporter.step("Get owner ID of another bucket"):
+            bucket_acl = s3_client.get_bucket_acl(another_bucket)
+            another_bucket_owner = bucket_acl["Owner"]["DisplayName"]
+
+        with reporter.step("Try patch object with x-amz-expected-bucket-owner header and get exception"):
+            with pytest.raises(Exception, match="AccessDenied"):
+                s3_http_client.patch_object(
+                    bucket,
+                    original_object,
+                    b"blablabla",
+                    "bytes 10-18/*",
+                    x_amz_expected_bucket_owner=another_bucket_owner,
+                )
+
+    @allure.title(
+        "[NEGATIVE] Patch cannot be applied with invalid Content-Range header "
+        "(range={patch_range}, s3_client={s3_client}, object_size={object_size}, policy={placement_policy})"
+    )
+    @pytest.mark.parametrize(
+        "patch_range",
+        # String "object" denotes size of object.
+        ["object+100:200", "object+10:object+16", "-1:1", "20:100", "0:2", f"0:{FIVE_GIGABYTES}", "0:0"],
+    )
+    def test_patch_with_invalid_content_range(
+        self,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        original_object: str,
+        object_size: ObjectSize,
+        patch_range: str,
+    ):
+        content_range = s3_helper.get_range_relative_to_object(patch_range, object_size.value)
+        with reporter.step("Try patch object with invalid Content-Range header and get exception"):
+            with pytest.raises(Exception, match="InvalidRange"):
+                s3_http_client.patch_object(bucket, original_object, b"content", content_range)
+
+    @allure.title(
+        "[NEGATIVE] Patch cannot be applied without Content-Range header "
+        "(s3_client={s3_client}, object_size={object_size}, policy={placement_policy})"
+    )
+    def test_patch_without_content_range(self, s3_http_client: S3HttpClient, bucket: str, original_object: str):
+        with reporter.step("Try patch object without Content-Range header and get exception"):
+            with pytest.raises(Exception, match="MissingContentRange"):
+                s3_http_client.patch_object(bucket, original_object, b"content", None)
+
+    @allure.title(
+        "[NEGATIVE] Patch cannot be applied without Content-Length header "
+        "(s3_client={s3_client}, object_size={object_size}, policy={placement_policy})"
+    )
+    def test_patch_without_content_length(
+        self,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        original_object: str,
+        node_under_test: ClusterNode,
+    ):
+        with reporter.step("Generate headers that comply with AWS specification"):
+            data = "content"
+            url = f"{self.cluster.default_s3_gate_endpoint}/{bucket}/{original_object}"
+            host = self.cluster.default_s3_gate_endpoint[8:]
+            headers = {"Host": host, "Url": url, "Content-Range": "bytes 0-6/*"}
+            headers = dict(s3_http_client._create_aws_request("PATCH", url, headers, data).headers)
+            headers.pop("Content-Length", None)
+
+        with reporter.step("Try patch object without Content-Length header and get exception"):
+            curl = GenericCli("curl", node_under_test.host)
+            request = f" {url} -X PATCH"
+
+            for header, value in headers.items():
+                request += f" -H '{header}: {value}'"
+
+            # Remove Content-Length header
+            # *Header without a value means to CURL that it should not be inserted into the request.
+            request += " -H 'Content-Length:'"
+            request += f" -d '{data}' -k"
+
+            response = curl(request, shell=LocalShell(), options=CommandOptions(check=False))
+            assert "MissingContentLength" in response.stdout, response.stdout
+
+    @allure.title("[NEGATIVE] Patch cannot be applied to non-existent bucket")
+    def test_patch_non_existent_bucket(self, s3_http_client: S3HttpClient):
+        with reporter.step("Try patch object in non-existent bucket and get exception"):
+            with pytest.raises(Exception, match="NoSuchBucket"):
+                s3_http_client.patch_object("fake-bucket", unique_name("object-"), b"content", "bytes 0-6/*")
+
+    @allure.title("[NEGATIVE] Patch cannot be applied to non-existent object (s3_client={s3_client}, policy={placement_policy})")
+    def test_patch_non_existent_object(self, s3_http_client: S3HttpClient, bucket: str):
+        with reporter.step("Try patch non-existent object and get exception"):
+            with pytest.raises(Exception, match="NoSuchKey"):
+                s3_http_client.patch_object(bucket, "fake-object", b"content", "bytes 0-6/*")
+
+    @allure.title("Patch object in versioned bucket (s3_client={s3_client}, object_size={object_size}, policy={placement_policy})")
+    @pytest.mark.parametrize("versioning_status", [VersioningStatus.ENABLED], indirect=True)
+    def test_patch_object_in_versioned_bucket(
+        self,
+        s3_client: S3ClientWrapper,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        object_size: ObjectSize,
+    ):
+        patch_ranges = ["0:35", "40:49", "object-100:object", "object:object+231"]
+
+        with reporter.step("Generate original object"):
+            original_file = generate_file(object_size.value)
+            original_key = s3_helper.object_key_from_file_path(original_file)
+
+        with reporter.step("Put object"):
+            version = s3_client.put_object(bucket, original_file, original_key)
+            expected_versions = {version}
+
+        with reporter.step("Patch versioned object"):
+            for rng in patch_ranges:
+                start, end = s3_helper.get_range_relative_to_object(rng, object_size=object_size.value, int_values=True)
+                content_size = end - start + 1
+                content_range = f"bytes {start}-{end}/*"
+
+                with reporter.step(f"Generate payload object of {content_size} bytes"):
+                    content_file = generate_file(content_size)
+
+                with reporter.step(f"Patch object and get new version"):
+                    response = s3_http_client.patch_object(bucket, original_key, content_file, content_range, version_id=version)
+                    version = response["VersionId"]
+                    expected_versions.add(version)
+
+                with reporter.step(f"Get patched part of object and make sure it has changed correctly"):
+                    got_part = s3_client.get_object(bucket, original_key, version_id=version, object_range=(start, end))
+                    assert get_file_hash(got_part) == get_file_hash(content_file), "Expected content hash did not match actual content hash"
+
+        with reporter.step("Check that all expected versions are in bucket"):
+            got_versions = {
+                version.get("VersionId") for version in s3_client.list_objects_versions(bucket) if version.get("Key") == original_key
+            }
+            assert expected_versions == got_versions, f"Expected versions of object are missing from bucket: {expected_versions}"
+
+    @allure.title("Patch multipart object (range={patch_range}, s3_client={s3_client}, policy={placement_policy})")
+    @pytest.mark.parametrize("patch_range", ["0:part-1", "part:part*2-1", "part-100:part*2+200", "object-part-1:object"])
+    def test_s3_patch_multipart_object(
+        self,
+        s3_client: S3ClientWrapper,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+        patch_range: str,
+    ):
+        parts_count = 5
+        parts = []
+        original_size = PART_SIZE_FOR_MULTIPART * parts_count
+
+        with reporter.step("Generate original object and split it into parts"):
+            original_file = generate_file(original_size)
+            file_parts = split_file(original_file, parts_count)
+            object_key = s3_helper.object_key_from_file_path(original_file)
+
+        start, end = s3_helper.get_range_relative_to_object(
+            patch_range, object_size=original_size, part_size=PART_SIZE_FOR_MULTIPART, int_values=True
+        )
+        content_size = end - start + 1
+        content_range = f"bytes {start}-{end}/*"
+
+        with reporter.step("Generate payload object"):
+            content_file = generate_file(content_size)
+
+        with reporter.step("Create multipart and upload parts"):
+            upload_id = s3_client.create_multipart_upload(bucket, object_key)
+            for part_id, file_path in enumerate(file_parts, start=1):
+                etag = s3_client.upload_part(bucket, object_key, upload_id, part_id, file_path)
+                parts.append((part_id, etag))
+
+        with reporter.step("Check all parts are visible in bucket"):
+            got_parts = s3_client.list_parts(bucket, object_key, upload_id)
+            assert len(got_parts) == len(file_parts), f"Expected {parts_count} parts, got:\n{got_parts}"
+
+        with reporter.step("Complete multipart upload"):
+            s3_client.complete_multipart_upload(bucket, object_key, upload_id, parts)
+
+        with reporter.step("Patch multipart object"):
+            s3_http_client.patch_object(bucket, object_key, content_file, content_range, timeout=200)
+
+        with reporter.step("Get patched part of object and make sure it has changed correctly"):
+            got_part = s3_client.get_object(bucket, object_key, object_range=(start, end))
+            assert get_file_hash(got_part) == get_file_hash(content_file), "Expected content hash did not match actual content hash"
+
+    @allure.title("Patch multipart object in versioned bucket (s3_client={s3_client}, policy={placement_policy})")
+    @pytest.mark.parametrize("versioning_status", [VersioningStatus.ENABLED], indirect=True)
+    def test_s3_patch_multipart_object_in_versioned_bucket(
+        self,
+        s3_client: S3ClientWrapper,
+        s3_http_client: S3HttpClient,
+        bucket: str,
+    ):
+        parts = []
+        parts_count = 5
+        original_size = PART_SIZE_FOR_MULTIPART * parts_count
+        patch_ranges = ["0:part-1", "part:part*2-1", "part-100:part*2+200", "object-part-1:object"]
+
+        with reporter.step("Generate original object and split it into parts"):
+            original_file = generate_file(original_size)
+            original_key = s3_helper.object_key_from_file_path(original_file)
+            file_parts = split_file(original_file, parts_count)
+
+        with reporter.step("Create multipart and upload parts"):
+            upload_id = s3_client.create_multipart_upload(bucket, original_key)
+            for part_id, file_path in enumerate(file_parts, start=1):
+                etag = s3_client.upload_part(bucket, original_key, upload_id, part_id, file_path)
+                parts.append((part_id, etag))
+
+        with reporter.step("Check all parts are visible in bucket"):
+            got_parts = s3_client.list_parts(bucket, original_key, upload_id)
+            assert len(got_parts) == len(file_parts), f"Expected {parts_count} parts, got:\n{got_parts}"
+
+        with reporter.step("Complete multipart upload"):
+            response = s3_client.complete_multipart_upload(bucket, original_key, upload_id, parts)
+            version = response["VersionId"]
+            expected_versions = {version}
+
+        with reporter.step("Patch versioned multipart object"):
+            for rng in patch_ranges:
+                start, end = s3_helper.get_range_relative_to_object(
+                    rng, object_size=original_size, part_size=PART_SIZE_FOR_MULTIPART, int_values=True
+                )
+                content_size = end - start + 1
+                content_range = f"bytes {start}-{end}/*"
+
+                with reporter.step("Generate payload object"):
+                    content_file = generate_file(content_size)
+
+                with reporter.step("Patch multipart object and get new version"):
+                    response = s3_http_client.patch_object(
+                        bucket, original_key, content_file, content_range, version_id=version, timeout=200
+                    )
+                    version = response["VersionId"]
+                    expected_versions.add(version)
+
+                with reporter.step("Get patched part of object and make sure it has changed correctly"):
+                    got_part = s3_client.get_object(bucket, original_key, version_id=version, object_range=(start, end))
+                    assert get_file_hash(got_part) == get_file_hash(content_file), "Expected content hash did not match actual content hash"
+
+        with reporter.step("Check that all expected versions are in bucket"):
+            got_versions = {
+                version.get("VersionId") for version in s3_client.list_objects_versions(bucket) if version.get("Key") == original_key
+            }
+            assert expected_versions == got_versions, f"Expected versions of object are missing from bucket: {expected_versions}"
+
+    # TODO: Negative scenario for SSE objects is postponed for now.

pytest_tests/testsuites/services/s3_gate/test_s3_locking.py

@@ -4,8 +4,8 @@ from datetime import datetime, timedelta
 import allure
 import pytest
 from frostfs_testlib import reporter
-from frostfs_testlib.s3 import S3ClientWrapper
-from frostfs_testlib.steps.s3 import s3_helper
+from frostfs_testlib.clients import S3ClientWrapper
+from frostfs_testlib.steps import s3_helper
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
 from frostfs_testlib.utils.file_utils import generate_file, generate_file_with_content
 

pytest_tests/testsuites/services/s3_gate/test_s3_multipart.py

@@ -1,10 +1,9 @@
 import allure
 import pytest
 from frostfs_testlib import reporter
-from frostfs_testlib.s3 import S3ClientWrapper, VersioningStatus
-from frostfs_testlib.s3.interfaces import BucketContainerResolver
+from frostfs_testlib.clients.s3 import BucketContainerResolver, S3ClientWrapper, VersioningStatus
+from frostfs_testlib.steps import s3_helper
 from frostfs_testlib.steps.cli.container import list_objects
-from frostfs_testlib.steps.s3 import s3_helper
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase

pytest_tests/testsuites/services/s3_gate/test_s3_object.py

@@ -8,11 +8,12 @@ from typing import Literal
 import allure
 import pytest
 from frostfs_testlib import reporter
+from frostfs_testlib.clients import AwsCliClient, S3ClientWrapper
+from frostfs_testlib.clients.s3 import VersioningStatus
 from frostfs_testlib.resources.common import ASSETS_DIR, DEFAULT_WALLET_PASS
 from frostfs_testlib.resources.error_patterns import S3_BUCKET_DOES_NOT_ALLOW_ACL, S3_MALFORMED_XML_REQUEST
 from frostfs_testlib.resources.s3_acl_grants import PRIVATE_GRANTS
-from frostfs_testlib.s3 import AwsCliClient, S3ClientWrapper, VersioningStatus
-from frostfs_testlib.steps.s3 import s3_helper
+from frostfs_testlib.steps import s3_helper
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
 from frostfs_testlib.testing.test_control import expect_not_raises
 from frostfs_testlib.utils import wallet_utils

pytest_tests/testsuites/services/s3_gate/test_s3_policy.py

@@ -4,9 +4,8 @@ import allure
 import pytest
 from botocore.exceptions import ClientError
 from frostfs_testlib import reporter
-from frostfs_testlib.s3 import S3ClientWrapper, VersioningStatus
-from frostfs_testlib.s3.interfaces import BucketContainerResolver
-from frostfs_testlib.steps.s3 import s3_helper
+from frostfs_testlib.clients.s3 import BucketContainerResolver, S3ClientWrapper, VersioningStatus
+from frostfs_testlib.steps import s3_helper
 from frostfs_testlib.steps.storage_policy import get_simple_object_copies
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
 from frostfs_testlib.storage.dataclasses.wallet import WalletInfo

pytest_tests/testsuites/services/s3_gate/test_s3_tagging.py

@@ -5,8 +5,8 @@ from typing import Tuple
 import allure
 import pytest
 from frostfs_testlib import reporter
-from frostfs_testlib.s3 import S3ClientWrapper
-from frostfs_testlib.steps.s3 import s3_helper
+from frostfs_testlib.clients import S3ClientWrapper
+from frostfs_testlib.steps import s3_helper
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
 from frostfs_testlib.utils.file_utils import generate_file
 

pytest_tests/testsuites/services/s3_gate/test_s3_versioning.py

@@ -3,8 +3,8 @@ import os
 import allure
 import pytest
 from frostfs_testlib import reporter
-from frostfs_testlib.s3 import S3ClientWrapper, VersioningStatus
-from frostfs_testlib.steps.s3 import s3_helper
+from frostfs_testlib.clients.s3 import S3ClientWrapper, VersioningStatus
+from frostfs_testlib.steps import s3_helper
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
 from frostfs_testlib.utils.file_utils import generate_file, generate_file_with_content, get_file_content
 

pytest_tests/testsuites/session_token/conftest.py

@@ -7,8 +7,8 @@ from frostfs_testlib.utils import string_utils
 
 
 @pytest.fixture(scope="module")
-def owner_wallet(default_wallet: WalletInfo) -> WalletInfo:
-    return default_wallet
+def owner_wallet(wallet: WalletInfo) -> WalletInfo:
+    return wallet
 
 
 @pytest.fixture(scope="module")

pytest_tests/testsuites/session_token/test_object_session_token.py

@@ -3,8 +3,8 @@ import random
 import allure
 import pytest
 from frostfs_testlib import reporter
+from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
 from frostfs_testlib.resources.error_patterns import SESSION_NOT_FOUND
-from frostfs_testlib.steps.cli.container import create_container
 from frostfs_testlib.steps.cli.object import delete_object, put_object, put_object_to_random_node
 from frostfs_testlib.steps.session_token import create_session_token
 from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
@@ -12,13 +12,17 @@ from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
 from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 from frostfs_testlib.utils.file_utils import generate_file
 
+from ...helpers.container_creation import create_container_with_ape
+from ...helpers.container_request import APE_OWNER_ALLOW_ALL, ContainerRequest
+
 
 @pytest.mark.nightly
 @pytest.mark.sanity
 @pytest.mark.session_token
 class TestDynamicObjectSession(ClusterTestBase):
     @allure.title("Object Operations with Session Token (obj_size={object_size})")
-    def test_object_session_token(self, default_wallet: WalletInfo, object_size: ObjectSize):
+    @pytest.mark.parametrize("user_tag", ["TestDynamicObjectSession"], indirect=True)  # provide dedicated user with no APE side-policies
+    def test_object_session_token(self, wallet: WalletInfo, frostfs_cli: FrostfsCli, object_size: ObjectSize):
         """
         Test how operations over objects are executed with a session token
 
@@ -32,17 +36,14 @@ class TestDynamicObjectSession(ClusterTestBase):
             with a session token
         """
 
-        with reporter.step("Init wallet"):
-            wallet = default_wallet
-
         with reporter.step("Nodes Settlements"):
             session_token_node, container_node, non_container_node = random.sample(self.cluster.storage_nodes, 3)
 
         with reporter.step("Create Session Token"):
             session_token = create_session_token(
                 shell=self.shell,
-                owner=default_wallet.get_address(),
-                wallet=default_wallet,
+                owner=wallet.get_address(),
+                wallet=wallet,
                 rpc_endpoint=session_token_node.get_rpc_endpoint(),
             )
 
@@ -54,11 +55,13 @@ class TestDynamicObjectSession(ClusterTestBase):
                 f'AS LOC_{locode}_PLACE FILTER "UN-LOCODE" '
                 f'EQ "{un_locode}" AS LOC_{locode}'
             )
-            cid = create_container(
+            cid = create_container_with_ape(
+                ContainerRequest(placement_policy, APE_OWNER_ALLOW_ALL),
+                frostfs_cli,
                 wallet,
-                shell=self.shell,
-                endpoint=self.cluster.default_rpc_endpoint,
-                rule=placement_policy,
+                self.shell,
+                self.cluster,
+                self.cluster.default_rpc_endpoint,
             )
 
         with reporter.step("Put Objects"):

pytest_tests/testsuites/session_token/test_static_object_session_token.py

@@ -3,9 +3,9 @@ import logging
 import allure
 import pytest
 from frostfs_testlib import reporter
+from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
 from frostfs_testlib.resources.error_patterns import EXPIRED_SESSION_TOKEN, MALFORMED_REQUEST, OBJECT_ACCESS_DENIED, OBJECT_NOT_FOUND
 from frostfs_testlib.shell import Shell
-from frostfs_testlib.steps.cli.container import create_container
 from frostfs_testlib.steps.cli.object import (
     delete_object,
     get_object,
@@ -38,16 +38,24 @@ from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
 from frostfs_testlib.testing.test_control import expect_not_raises
 from frostfs_testlib.utils.file_utils import generate_file
 
+from ...helpers.container_creation import create_containers_with_ape
+from ...helpers.container_request import OWNER_ALLOW_ALL, MultipleContainersRequest
+
 logger = logging.getLogger("NeoLogger")
 
 RANGE_OFFSET_FOR_COMPLEX_OBJECT = 200
 
 
 @pytest.fixture(scope="module")
-def storage_containers(owner_wallet: WalletInfo, client_shell: Shell, cluster: Cluster) -> list[str]:
-    cid = create_container(owner_wallet, shell=client_shell, endpoint=cluster.default_rpc_endpoint)
-    other_cid = create_container(owner_wallet, shell=client_shell, endpoint=cluster.default_rpc_endpoint)
-    yield [cid, other_cid]
+def storage_containers(owner_wallet: WalletInfo, frostfs_cli: FrostfsCli, client_shell: Shell, cluster: Cluster) -> list[str]:
+    return create_containers_with_ape(
+        frostfs_cli,
+        owner_wallet,
+        client_shell,
+        cluster,
+        cluster.default_rpc_endpoint,
+        MultipleContainersRequest([OWNER_ALLOW_ALL, OWNER_ALLOW_ALL]),
+    )
 
 
 @pytest.fixture(
@@ -135,6 +143,7 @@ def static_sessions(
 
 @pytest.mark.nightly
 @pytest.mark.static_session
+@pytest.mark.parametrize("user_tag", ["TestObjectStaticSession"], indirect=True)  # provide dedicated user with no APE side-policies
 class TestObjectStaticSession(ClusterTestBase):
     @allure.title("Read operations with static session (method={method_under_test.__name__}, obj_size={object_size})")
     @pytest.mark.parametrize(