frostfs-testcases/robot/resources/lib/python_keywords/object_access.py

from typing import Optional

import allure

from grpc_responses import OBJECT_ACCESS_DENIED, error_matches_status
from python_keywords.neofs_verbs import (delete_object, get_object, get_range, get_range_hash, head_object, put_object,
                                         search_object)
from python_keywords.utility_keywords import get_file_hash

OPERATION_ERROR_TYPE = RuntimeError


def can_get_object(wallet: str, cid: str, oid: str, file_name: str, bearer: Optional[str] = None,
                   wallet_config: Optional[str] = None, xhdr: Optional[dict] = None
                   ) -> bool:
    with allure.step('Try get object from container'):
        try:
            got_file_path = get_object(wallet, cid, oid, bearer_token=bearer, wallet_config=wallet_config, xhdr=xhdr)
        except OPERATION_ERROR_TYPE as err:
            assert error_matches_status(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, 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(wallet, file_name, cid, bearer=bearer, wallet_config=wallet_config, xhdr=xhdr,
                       attributes=attributes)
        except OPERATION_ERROR_TYPE as err:
            assert error_matches_status(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, bearer: Optional[str] = None,
                      wallet_config: Optional[str] = None, xhdr: Optional[dict] = None
                      ) -> bool:
    with allure.step('Try delete object from container'):
        try:
            delete_object(wallet, cid, oid, bearer=bearer, wallet_config=wallet_config, xhdr=xhdr)
        except OPERATION_ERROR_TYPE as err:
            assert error_matches_status(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, 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_token=bearer, wallet_config=wallet_config, xhdr=xhdr)
        except OPERATION_ERROR_TYPE as err:
            assert error_matches_status(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, 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)
        except OPERATION_ERROR_TYPE as err:
            assert error_matches_status(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, 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_token=bearer, range_cut='0:10', wallet_config=wallet_config,
                           xhdr=xhdr)
        except OPERATION_ERROR_TYPE as err:
            assert error_matches_status(err, OBJECT_ACCESS_DENIED), f'Expected {err} to match {OBJECT_ACCESS_DENIED}'
            return False
    return True


def can_search_object(wallet: str, cid: 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)
        except OPERATION_ERROR_TYPE as err:
            assert error_matches_status(err, OBJECT_ACCESS_DENIED), f'Expected {err} to match {OBJECT_ACCESS_DENIED}'
            return False
        if oid:
            return oid in oids
    return True