frostfs-testcases/pytest_tests/helpers/object_access.py

238 lines
6.5 KiB
Python
Raw Normal View History

from typing import Optional
import allure
from frostfs_testlib.resources.common import OBJECT_ACCESS_DENIED
from frostfs_testlib.shell import Shell
from frostfs_testlib.utils import string_utils
from pytest_tests.helpers.cluster import Cluster
from pytest_tests.helpers.file_helper import get_file_hash
from pytest_tests.helpers.frostfs_verbs import (
delete_object,
get_object_from_random_node,
get_range,
get_range_hash,
head_object,
put_object_to_random_node,
search_object,
)
OPERATION_ERROR_TYPE = RuntimeError
def can_get_object(
wallet: str,
cid: str,
oid: str,
file_name: str,
shell: Shell,
cluster: Cluster,
bearer: Optional[str] = None,
wallet_config: Optional[str] = None,
xhdr: Optional[dict] = None,
) -> bool:
with allure.step("Try get object from pytest_tests.helpers.container"):
try:
got_file_path = get_object_from_random_node(
wallet,
cid,
oid,
bearer=bearer,
wallet_config=wallet_config,
xhdr=xhdr,
shell=shell,
cluster=cluster,
)
except OPERATION_ERROR_TYPE as err:
assert string_utils.is_str_match_pattern(
err, OBJECT_ACCESS_DENIED
), f"Expected {err} to match {OBJECT_ACCESS_DENIED}"
return False
assert get_file_hash(file_name) == get_file_hash(got_file_path)
return True
def can_put_object(
wallet: str,
cid: str,
file_name: str,
shell: Shell,
cluster: Cluster,
bearer: Optional[str] = None,
wallet_config: Optional[str] = None,
xhdr: Optional[dict] = None,
attributes: Optional[dict] = None,
) -> bool:
with allure.step("Try put object to container"):
try:
put_object_to_random_node(
wallet,
file_name,
cid,
bearer=bearer,
wallet_config=wallet_config,
xhdr=xhdr,
attributes=attributes,
shell=shell,
cluster=cluster,
)
except OPERATION_ERROR_TYPE as err:
assert string_utils.is_str_match_pattern(
err, OBJECT_ACCESS_DENIED
), f"Expected {err} to match {OBJECT_ACCESS_DENIED}"
return False
return True
def can_delete_object(
wallet: str,
cid: str,
oid: str,
shell: Shell,
endpoint: str,
bearer: Optional[str] = None,
wallet_config: Optional[str] = None,
xhdr: Optional[dict] = None,
) -> bool:
with allure.step("Try delete object from pytest_tests.helpers.container"):
try:
delete_object(
wallet,
cid,
oid,
bearer=bearer,
wallet_config=wallet_config,
xhdr=xhdr,
shell=shell,
endpoint=endpoint,
)
except OPERATION_ERROR_TYPE as err:
assert string_utils.is_str_match_pattern(
err, OBJECT_ACCESS_DENIED
), f"Expected {err} to match {OBJECT_ACCESS_DENIED}"
return False
return True
def can_get_head_object(
wallet: str,
cid: str,
oid: str,
shell: Shell,
endpoint: str,
bearer: Optional[str] = None,
wallet_config: Optional[str] = None,
xhdr: Optional[dict] = None,
) -> bool:
with allure.step("Try get head of object"):
try:
head_object(
wallet,
cid,
oid,
bearer=bearer,
wallet_config=wallet_config,
xhdr=xhdr,
shell=shell,
endpoint=endpoint,
)
except OPERATION_ERROR_TYPE as err:
assert string_utils.is_str_match_pattern(
err, OBJECT_ACCESS_DENIED
), f"Expected {err} to match {OBJECT_ACCESS_DENIED}"
return False
return True
def can_get_range_of_object(
wallet: str,
cid: str,
oid: str,
shell: Shell,
endpoint: str,
bearer: Optional[str] = None,
wallet_config: Optional[str] = None,
xhdr: Optional[dict] = None,
) -> bool:
with allure.step("Try get range of object"):
try:
get_range(
wallet,
cid,
oid,
bearer=bearer,
range_cut="0:10",
wallet_config=wallet_config,
xhdr=xhdr,
shell=shell,
endpoint=endpoint,
)
except OPERATION_ERROR_TYPE as err:
assert string_utils.is_str_match_pattern(
err, OBJECT_ACCESS_DENIED
), f"Expected {err} to match {OBJECT_ACCESS_DENIED}"
return False
return True
def can_get_range_hash_of_object(
wallet: str,
cid: str,
oid: str,
shell: Shell,
endpoint: str,
bearer: Optional[str] = None,
wallet_config: Optional[str] = None,
xhdr: Optional[dict] = None,
) -> bool:
with allure.step("Try get range hash of object"):
try:
get_range_hash(
wallet,
cid,
oid,
bearer=bearer,
range_cut="0:10",
wallet_config=wallet_config,
xhdr=xhdr,
shell=shell,
endpoint=endpoint,
)
except OPERATION_ERROR_TYPE as err:
assert string_utils.is_str_match_pattern(
err, OBJECT_ACCESS_DENIED
), f"Expected {err} to match {OBJECT_ACCESS_DENIED}"
return False
return True
def can_search_object(
wallet: str,
cid: str,
shell: Shell,
endpoint: str,
oid: Optional[str] = None,
bearer: Optional[str] = None,
wallet_config: Optional[str] = None,
xhdr: Optional[dict] = None,
) -> bool:
with allure.step("Try search object in container"):
try:
oids = search_object(
wallet,
cid,
bearer=bearer,
wallet_config=wallet_config,
xhdr=xhdr,
shell=shell,
endpoint=endpoint,
)
except OPERATION_ERROR_TYPE as err:
assert string_utils.is_str_match_pattern(
err, OBJECT_ACCESS_DENIED
), f"Expected {err} to match {OBJECT_ACCESS_DENIED}"
return False
if oid:
return oid in oids
return True