pyproject.toml

@@ -57,6 +57,9 @@ frostfs-http = "frostfs_testlib.storage.dataclasses.frostfs_services:HTTPGate"
 neo-go = "frostfs_testlib.storage.dataclasses.frostfs_services:MorphChain"
 frostfs-ir = "frostfs_testlib.storage.dataclasses.frostfs_services:InnerRing"
 
+[project.entry-points."frostfs.testlib.credentials_providers"]
+authmate = "frostfs_testlib.credentials.authmate_s3:AuthmateS3CredentialsProvider"
+
 [tool.isort]
 profile = "black"
 src_paths = ["src", "tests"]

src/frostfs_testlib/credentials/authmate_s3.py

@@ -0,0 +1,49 @@
+import re
+from datetime import datetime
+
+from frostfs_testlib import reporter
+from frostfs_testlib.cli import FrostfsAuthmate
+from frostfs_testlib.credentials.interfaces import S3CredentialsProvider
+from frostfs_testlib.resources.cli import FROSTFS_AUTHMATE_EXEC
+from frostfs_testlib.shell import Shell
+from frostfs_testlib.steps.cli.container import list_containers
+from frostfs_testlib.storage.cluster import Cluster, ClusterNode
+from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
+
+
+class AuthmateS3CredentialsProvider(S3CredentialsProvider):
+    @reporter.step("Init S3 Credentials using Authmate CLI")
+    def provide(self, cluster_node: ClusterNode) -> tuple[str, str]:
+        cluster: Cluster = self.stash["cluster"]
+        shell: Shell = self.stash["shell"]
+        wallet: WalletInfo = self.stash["wallet"]
+        endpoint = cluster_node.storage_node.get_rpc_endpoint()
+
+        gate_public_keys = [s3gate.get_wallet_public_key() for s3gate in cluster.s3_gates]
+        # unique short bucket name
+        bucket = f"bucket_{hex(int(datetime.now().timestamp()*1000000))}"
+
+        frostfs_authmate: FrostfsAuthmate = FrostfsAuthmate(shell, FROSTFS_AUTHMATE_EXEC)
+        issue_secret_output = frostfs_authmate.secret.issue(
+            wallet=wallet.path,
+            peer=endpoint,
+            gate_public_key=gate_public_keys,
+            wallet_password=wallet.password,
+            container_policy=self.stash.get("location_constraints"),
+            container_friendly_name=bucket,
+        ).stdout
+
+        aws_access_key_id = str(
+            re.search(r"access_key_id.*:\s.(?P<aws_access_key_id>\w*)", issue_secret_output).group("aws_access_key_id")
+        )
+        aws_secret_access_key = str(
+            re.search(r"secret_access_key.*:\s.(?P<aws_secret_access_key>\w*)", issue_secret_output).group(
+                "aws_secret_access_key"
+            )
+        )
+        cid = str(re.search(r"container_id.*:\s.(?P<container_id>\w*)", issue_secret_output).group("container_id"))
+
+        containers_list = list_containers(wallet.path, shell, endpoint)
+        assert cid in containers_list, f"Expected cid {cid} in {containers_list}"
+
+        return aws_access_key_id, aws_secret_access_key

src/frostfs_testlib/credentials/interfaces.py

@@ -0,0 +1,25 @@
+from abc import abstractmethod
+
+from frostfs_testlib.plugins import load_plugin
+from frostfs_testlib.storage.cluster import ClusterNode
+
+
+class S3CredentialsProvider(object):
+    stash: dict
+
+    def __init__(self, stash: dict) -> None:
+        self.stash = stash
+
+    @abstractmethod
+    def provide(self, cluster_node: ClusterNode) -> tuple[str, str]:
+        raise NotImplementedError("Directly called abstract class?")
+
+
+class CredentialsProvider(object):
+    stash: dict
+    S3: S3CredentialsProvider
+
+    def __init__(self, s3_plugin_name: str) -> None:
+        self.stash = {}
+        s3cls = load_plugin("frostfs.testlib.credentials_providers", s3_plugin_name)
+        self.S3 = s3cls(self.stash)

src/frostfs_testlib/hosting/config.py

@@ -62,6 +62,7 @@ class HostConfig:
     plugin_name: str
     healthcheck_plugin_name: str
     address: str
+    s3_creds_plugin_name: str = field(default="authmate")
     services: list[ServiceConfig] = field(default_factory=list)
     clis: list[CLIConfig] = field(default_factory=list)
     attributes: dict[str, str] = field(default_factory=dict)

src/frostfs_testlib/steps/s3/s3_helper.py

@@ -1,25 +1,15 @@
-import json
 import logging
 import os
-import re
-import uuid
 from datetime import datetime, timedelta
 from typing import Optional
 
 from dateutil.parser import parse
 
 from frostfs_testlib import reporter
-from frostfs_testlib.cli import FrostfsAuthmate
-from frostfs_testlib.resources.cli import FROSTFS_AUTHMATE_EXEC
-from frostfs_testlib.resources.common import CREDENTIALS_CREATE_TIMEOUT
 from frostfs_testlib.s3 import S3ClientWrapper, VersioningStatus
-from frostfs_testlib.shell import CommandOptions, InteractiveInput, Shell
+from frostfs_testlib.shell import Shell
-from frostfs_testlib.shell.interfaces import SshCredentials
 from frostfs_testlib.steps.cli.container import search_container_by_name, search_nodes_with_container
 from frostfs_testlib.storage.cluster import Cluster, ClusterNode
-from frostfs_testlib.storage.dataclasses.frostfs_services import S3Gate
-from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
-from frostfs_testlib.utils.cli_utils import _run_with_passwd
 
 logger = logging.getLogger("NeoLogger")
 
@@ -161,43 +151,6 @@ def assert_s3_acl(acl_grants: list, permitted_users: str):
                 logger.error("FULL_CONTROL is given to All Users")
 
 
-@reporter.step("Init S3 Credentials")
-def init_s3_credentials(
-    wallet: WalletInfo,
-    shell: Shell,
-    cluster: Cluster,
-    policy: Optional[dict] = None,
-    s3gates: Optional[list[S3Gate]] = None,
-    container_placement_policy: Optional[str] = None,
-):
-    gate_public_keys = []
-    bucket = str(uuid.uuid4())
-    if not s3gates:
-        s3gates = [cluster.s3_gates[0]]
-    for s3gate in s3gates:
-        gate_public_keys.append(s3gate.get_wallet_public_key())
-    frostfs_authmate_exec: FrostfsAuthmate = FrostfsAuthmate(shell, FROSTFS_AUTHMATE_EXEC)
-    issue_secret_output = frostfs_authmate_exec.secret.issue(
-        wallet=wallet.path,
-        peer=cluster.default_rpc_endpoint,
-        gate_public_key=gate_public_keys,
-        wallet_password=wallet.password,
-        container_policy=policy,
-        container_friendly_name=bucket,
-        container_placement_policy=container_placement_policy,
-    ).stdout
-    aws_access_key_id = str(
-        re.search(r"access_key_id.*:\s.(?P<aws_access_key_id>\w*)", issue_secret_output).group("aws_access_key_id")
-    )
-    aws_secret_access_key = str(
-        re.search(r"secret_access_key.*:\s.(?P<aws_secret_access_key>\w*)", issue_secret_output).group(
-            "aws_secret_access_key"
-        )
-    )
-    cid = str(re.search(r"container_id.*:\s.(?P<container_id>\w*)", issue_secret_output).group("container_id"))
-    return cid, aws_access_key_id, aws_secret_access_key
-
-
 @reporter.step("Delete bucket with all objects")
 def delete_bucket_with_objects(s3_client: S3ClientWrapper, bucket: str):
     versioning_status = s3_client.get_bucket_versioning_status(bucket)