Adapt http tests to new dynamic env

Signed-off-by: Evgeniy Zayats <[email protected]>

dynamic_env_pytest_tests/lib/helpers/storage_object_info.py

@@ -0,0 +1,25 @@
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class ObjectRef:
+    cid: str
+    oid: str
+
+
+@dataclass
+class LockObjectInfo(ObjectRef):
+    lifetime: Optional[int] = None
+    expire_at: Optional[int] = None
+
+
+@dataclass
+class StorageObjectInfo(ObjectRef):
+    size: Optional[int] = None
+    wallet_file_path: Optional[str] = None
+    file_path: Optional[str] = None
+    file_hash: Optional[str] = None
+    attributes: Optional[list[dict[str, str]]] = None
+    tombstone: Optional[str] = None
+    locks: Optional[list[LockObjectInfo]] = None

dynamic_env_pytest_tests/lib/http_gw/http_utils.py

@@ -0,0 +1,78 @@
+import random
+
+import allure
+import neofs_verbs
+from grpc_responses import OBJECT_NOT_FOUND, error_matches_status
+from neofs_testlib.env.env import StorageNode
+from neofs_testlib.shell import Shell
+from python_keywords.http_gate import assert_hashes_are_equal, get_via_http_gate
+from python_keywords.neofs_verbs import get_object
+
+
+def get_object_and_verify_hashes(
+    oid: str,
+    file_name: str,
+    wallet: str,
+    cid: str,
+    shell: Shell,
+    nodes: list[StorageNode],
+    endpoint: str,
+    object_getter=None,
+) -> None:
+    nodes_list = get_nodes_without_object(
+        wallet=wallet,
+        cid=cid,
+        oid=oid,
+        shell=shell,
+        nodes=nodes,
+    )
+    # for some reason we can face with case when nodes_list is empty due to object resides in all nodes
+    if nodes_list:
+        random_node = random.choice(nodes_list)
+    else:
+        random_node = random.choice(nodes)
+
+    object_getter = object_getter or get_via_http_gate
+
+    got_file_path = get_object(
+        wallet=wallet,
+        cid=cid,
+        oid=oid,
+        shell=shell,
+        endpoint=random_node.endpoint,
+    )
+    got_file_path_http = object_getter(cid=cid, oid=oid, endpoint=endpoint)
+
+    assert_hashes_are_equal(file_name, got_file_path, got_file_path_http)
+
+
+@allure.step("Get Nodes Without Object")
+def get_nodes_without_object(
+    wallet: str, cid: str, oid: str, shell: Shell, nodes: list[StorageNode]
+) -> list[StorageNode]:
+    """
+    The function returns list of nodes which do not store
+    the given object.
+    Args:
+         wallet (str): the path to the wallet on whose behalf
+                     we request the nodes
+         cid (str): ID of the container which store the object
+         oid (str): object ID
+         shell: executor for cli command
+    Returns:
+         (list): nodes which do not store the object
+    """
+    nodes_list = []
+    for node in nodes:
+        try:
+            res = neofs_verbs.head_object(
+                wallet, cid, oid, shell=shell, endpoint=node.endpoint, is_direct=True
+            )
+            if res is None:
+                nodes_list.append(node)
+        except Exception as err:
+            if error_matches_status(err, OBJECT_NOT_FOUND):
+                nodes_list.append(node)
+            else:
+                raise Exception(f"Got error {err} on head object command") from err
+    return nodes_list

dynamic_env_pytest_tests/pytest.ini

@@ -1,8 +1,8 @@
 [pytest]
 log_cli = 1
 log_cli_level = debug
-log_cli_format = %(asctime)s [%(levelname)4s] %(message)s
-log_format = %(asctime)s [%(levelname)4s] %(message)s
+log_cli_format = [%(threadName)s] %(asctime)s [%(levelname)4s] %(message)s 
+log_format = [%(threadName)s] %(asctime)s [%(levelname)4s] %(message)s [%(threadName)s]
 log_cli_date_format = %Y-%m-%d %H:%M:%S
 log_date_format = %H:%M:%S
 markers =

dynamic_env_pytest_tests/tests/services/http_gate/test_http_bearer.py

@@ -0,0 +1,140 @@
+import logging
+
+import allure
+import pytest
+from container import create_container
+from file_helper import generate_file
+from http_gate import upload_via_http_gate_curl
+from http_gw.http_utils import get_object_and_verify_hashes
+from neofs_env.neofs_env_test_base import NeofsEnvTestBase
+from python_keywords.acl import (
+    EACLAccess,
+    EACLOperation,
+    EACLRole,
+    EACLRule,
+    bearer_token_base64_from_file,
+    create_eacl,
+    form_bearertoken_file,
+    set_eacl,
+    sign_bearer,
+    wait_for_cache_expired,
+)
+from wellknown_acl import PUBLIC_ACL
+
+logger = logging.getLogger("NeoLogger")
+
+
+@pytest.mark.sanity
+@pytest.mark.http_gate
+class Test_http_bearer(NeofsEnvTestBase):
+    PLACEMENT_RULE = "REP 2 IN X CBF 1 SELECT 2 FROM * AS X"
+
+    @pytest.fixture(scope="class", autouse=True)
+    @allure.title("[Class/Autouse]: Prepare wallet and deposit")
+    def prepare_wallet(self, default_wallet):
+        Test_http_bearer.wallet = default_wallet
+
+    @pytest.fixture(scope="class")
+    def user_container(self) -> str:
+        return create_container(
+            wallet=self.wallet.path,
+            shell=self.shell,
+            endpoint=self.neofs_env.sn_rpc,
+            rule=self.PLACEMENT_RULE,
+            basic_acl=PUBLIC_ACL,
+        )
+
+    @pytest.fixture(scope="class")
+    def eacl_deny_for_others(self, user_container: str) -> None:
+        with allure.step(f"Set deny all operations for {EACLRole.OTHERS} via eACL"):
+            eacl = EACLRule(
+                access=EACLAccess.DENY, role=EACLRole.OTHERS, operation=EACLOperation.PUT
+            )
+            set_eacl(
+                self.wallet.path,
+                user_container,
+                create_eacl(user_container, eacl, shell=self.shell),
+                shell=self.shell,
+                endpoint=self.neofs_env.sn_rpc,
+            )
+            wait_for_cache_expired()
+
+    @pytest.fixture(scope="class")
+    def bearer_token_no_limit_for_others(self, user_container: str) -> str:
+        with allure.step(f"Create bearer token for {EACLRole.OTHERS} with all operations allowed"):
+            bearer = form_bearertoken_file(
+                self.wallet.path,
+                user_container,
+                [
+                    EACLRule(operation=op, access=EACLAccess.ALLOW, role=EACLRole.OTHERS)
+                    for op in EACLOperation
+                ],
+                shell=self.shell,
+                endpoint=self.neofs_env.sn_rpc,
+                sign=False,
+            )
+            bearer_signed = f"{bearer}_signed"
+            sign_bearer(
+                shell=self.shell,
+                wallet_path=self.wallet.path,
+                eacl_rules_file_from=bearer,
+                eacl_rules_file_to=bearer_signed,
+                json=False,
+            )
+            return bearer_token_base64_from_file(bearer_signed)
+
+    @allure.title(f"[negative] Put object without bearer token for {EACLRole.OTHERS}")
+    def test_unable_put_without_bearer_token(
+        self, simple_object_size: int, user_container: str, eacl_deny_for_others
+    ):
+        eacl_deny_for_others
+        upload_via_http_gate_curl(
+            cid=user_container,
+            filepath=generate_file(simple_object_size),
+            endpoint=f"http://{self.neofs_env.http_gw.address}",
+            error_pattern="access to object operation denied",
+        )
+
+    @pytest.mark.parametrize("bearer_type", ("header", "cookie"))
+    @pytest.mark.parametrize(
+        "object_size",
+        [pytest.lazy_fixture("simple_object_size"), pytest.lazy_fixture("complex_object_size")],
+        ids=["simple object", "complex object"],
+    )
+    def test_put_with_bearer_when_eacl_restrict(
+        self,
+        object_size: int,
+        bearer_type: str,
+        user_container: str,
+        eacl_deny_for_others,
+        bearer_token_no_limit_for_others: str,
+    ):
+        eacl_deny_for_others
+        bearer = bearer_token_no_limit_for_others
+        file_path = generate_file(object_size)
+        with allure.step(
+            f"Put object with bearer token for {EACLRole.OTHERS}, then get and verify hashes"
+        ):
+            headers = None
+            cookies = None
+            if bearer_type == "header":
+                headers = [f" -H 'Authorization: Bearer {bearer}'"]
+            if bearer_type == "cookie":
+                cookies = {"Bearer": bearer}
+
+            oid = upload_via_http_gate_curl(
+                cid=user_container,
+                filepath=file_path,
+                endpoint=f"http://{self.neofs_env.http_gw.address}",
+                headers=headers,
+                cookies=cookies,
+            )
+            get_object_and_verify_hashes(
+                oid=oid,
+                file_name=file_path,
+                wallet=self.wallet.path,
+                cid=user_container,
+                shell=self.shell,
+                nodes=self.neofs_env.storage_nodes,
+                endpoint=f"http://{self.neofs_env.http_gw.address}",
+            )