Add grpc lock tests

Signed-off-by: Andrey Berezin <a.berezin@yadro.com>

robot/resources/lib/python_keywords/complex_object_actions.py

@@ -29,6 +29,7 @@ def get_link_object(
     shell: Shell,
     bearer: str = "",
     wallet_config: str = WALLET_CONFIG,
+    is_direct: bool = True,
 ):
     """
     Args:
@@ -39,6 +40,8 @@ def get_link_object(
         shell: executor for cli command
         bearer (optional, str): path to Bearer token file
         wallet_config (optional, str): path to the neofs-cli config file
+        is_direct: send request directly to the node or not; this flag
+                   turns into `--ttl 1` key
     Returns:
         (str): Link Object ID
         When no Link Object ID is found after all Storage Nodes polling,
@@ -53,7 +56,7 @@ def get_link_object(
                 shell=shell,
                 endpoint=node,
                 is_raw=True,
-                is_direct=True,
+                is_direct=is_direct,
                 bearer=bearer,
                 wallet_config=wallet_config,
             )

robot/resources/lib/python_keywords/epoch.py

@@ -21,6 +21,16 @@ from utility import parse_time
 logger = logging.getLogger("NeoLogger")
 
 
+@allure.step("Ensure fresh epoch")
+def ensure_fresh_epoch(shell: Shell) -> int:
+    # ensure new fresh epoch to avoid epoch switch during test session
+    current_epoch = get_epoch(shell)
+    tick_epoch(shell)
+    epoch = get_epoch(shell)
+    assert epoch > current_epoch, "Epoch wasn't ticked"
+    return epoch
+
+
 @allure.step("Get Epoch")
 def get_epoch(shell: Shell):
     neogo = NeoGo(shell=shell, neo_go_exec_path=NEOGO_EXECUTABLE)

robot/resources/lib/python_keywords/neofs_verbs.py

@@ -272,6 +272,63 @@ def get_range(
     return range_file_path, content
 
 
+@allure.step("Lock Object")
+def lock_object(
+    wallet: str,
+    cid: str,
+    oid: str,
+    shell: Shell,
+    lifetime: Optional[int] = None,
+    expire_at: Optional[int] = None,
+    endpoint: Optional[str] = None,
+    address: Optional[str] = None,
+    bearer: Optional[str] = None,
+    session: Optional[str] = None,
+    wallet_config: Optional[str] = None,
+    ttl: Optional[int] = None,
+    xhdr: Optional[dict] = None,
+) -> str:
+    """
+    Lock object in container.
+
+    Args:
+        address: Address of wallet account.
+        bearer: File with signed JSON or binary encoded bearer token.
+        cid: Container ID.
+        oid: Object ID.
+        lifetime: Lock lifetime.
+        expire_at: Lock expiration epoch.
+        endpoint: Remote node address.
+        session: Path to a JSON-encoded container session token.
+        ttl: TTL value in request meta header (default 2).
+        wallet: WIF (NEP-2) string or path to the wallet or binary key.
+        xhdr: Dict with request X-Headers.
+
+    Returns:
+        Lock object ID
+    """
+
+    cli = NeofsCli(shell, NEOFS_CLI_EXEC, wallet_config or WALLET_CONFIG)
+    result = cli.object.lock(
+        rpc_endpoint=endpoint or NEOFS_ENDPOINT,
+        lifetime=lifetime,
+        expire_at=expire_at,
+        address=address,
+        wallet=wallet,
+        cid=cid,
+        oid=oid,
+        bearer=bearer,
+        xhdr=xhdr,
+        session=session,
+        ttl=ttl,
+    )
+
+    # splitting CLI output to lines and taking the penultimate line
+    id_str = result.stdout.strip().split("\n")[0]
+    oid = id_str.split(":")[1]
+    return oid.strip()
+
+
 @allure.step("Search object")
 def search_object(
     wallet: str,