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

@@ -1,16 +1,19 @@
 import json
 import logging
 import re
+import time
 from typing import List, Optional, Union
 
 from frostfs_testlib import reporter
 from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
 from frostfs_testlib.plugins import load_plugin
 from frostfs_testlib.resources.cli import CLI_DEFAULT_TIMEOUT
+from frostfs_testlib.resources.common import MORPH_BLOCK_TIME
 from frostfs_testlib.s3.interfaces import BucketContainerResolver
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode
+from frostfs_testlib.storage.dataclasses import ape
 from frostfs_testlib.storage.grpc_operations import interfaces
-from frostfs_testlib.utils import json_utils
+from frostfs_testlib.utils import datetime_utils, json_utils
 
 logger = logging.getLogger("NeoLogger")
 
@@ -93,6 +96,42 @@ class ContainerOperations(interfaces.ContainerInterface):
 
         return cid
 
+    @reporter.step("Create Container with APE")
+    def create_with_ape(
+        self,
+        endpoint: str,
+        nns_zone: Optional[str] = None,
+        nns_name: Optional[str] = None,
+        address: Optional[str] = None,
+        attributes: Optional[dict] = None,
+        basic_acl: Optional[str] = None,
+        await_mode: bool = False,
+        disable_timestamp: bool = False,
+        force: bool = False,
+        trace: bool = False,
+        name: Optional[str] = None,
+        nonce: Optional[str] = None,
+        policy: Optional[str] = None,
+        session: Optional[str] = None,
+        subnet: Optional[str] = None,
+        ttl: Optional[int] = None,
+        xhdr: Optional[dict] = None,
+        timeout: Optional[str] = CLI_DEFAULT_TIMEOUT,
+        ape_rules: List[ape.Rule] = None,
+    ) -> str:
+        with reporter.step("Create container"):
+            cid = self.create(
+                **{param: value for param, value in locals().items() if param not in ["self", "ape_rules"]},
+            )
+        if ape_rules:
+            with reporter.step("Apply APE rules for container"):
+                self._apply_ape_rules(cid, endpoint, ape_rules)
+
+            with reporter.step("Wait for one block"):
+                time.sleep(datetime_utils.parse_time(MORPH_BLOCK_TIME))
+
+        return cid
+
     @reporter.step("List Containers")
     def list(
         self,
@@ -298,6 +337,18 @@ class ContainerOperations(interfaces.ContainerInterface):
         with reporter.step(f"Return nodes - {nodes_list}"):
             return nodes_list
 
+    def _apply_ape_rules(self, cid: str, endpoint: str, ape_rules: List[ape.Rule]):
+        for ape_rule in ape_rules:
+            rule_str = ape_rule.as_string()
+            with reporter.step(f"Apply APE rule '{rule_str}' for container {cid}"):
+                self.cli.ape_manager.add(
+                    endpoint,
+                    ape_rule.chain_id,
+                    target_name=cid,
+                    target_type="container",
+                    rule=rule_str,
+                )
+
     @reporter.step("Resolve container by name")
     def resolve_container_by_name(name: str, node: ClusterNode):
         resolver_cls = load_plugin("frostfs.testlib.bucket_cid_resolver", node.host.config.product)

src/frostfs_testlib/storage/grpc_operations/interfaces.py

@@ -3,7 +3,7 @@ from typing import Any, List, Optional
 
 from frostfs_testlib.shell.interfaces import CommandResult
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode
-from frostfs_testlib.storage.constants import PlacementRule
+from frostfs_testlib.storage.dataclasses import ape
 from frostfs_testlib.storage.dataclasses.storage_object_info import Chunk, NodeNetmapInfo
 from frostfs_testlib.utils import file_utils
 
@@ -294,6 +294,36 @@ class ContainerInterface(ABC):
         """
         raise NotImplementedError("No implemethed method create")
 
+    @abstractmethod
+    def create_with_ape(
+        self,
+        endpoint: str,
+        nns_zone: Optional[str] = None,
+        nns_name: Optional[str] = None,
+        address: Optional[str] = None,
+        attributes: Optional[dict] = None,
+        basic_acl: Optional[str] = None,
+        await_mode: bool = False,
+        disable_timestamp: bool = False,
+        force: bool = False,
+        trace: bool = False,
+        name: Optional[str] = None,
+        nonce: Optional[str] = None,
+        policy: Optional[str] = None,
+        session: Optional[str] = None,
+        subnet: Optional[str] = None,
+        ttl: Optional[int] = None,
+        xhdr: Optional[dict] = None,
+        timeout: Optional[str] = None,
+        ape_rules: List[ape.Rule] = None,
+    ) -> str:
+        """
+        Create a new container and register it in the FrostFS.
+        It will be stored in the sidechain when the Inner Ring accepts it.
+        Add ape rules at each nodes
+        """
+        raise NotImplementedError("No implemethed method create")
+
     @abstractmethod
     def delete(
         self,