frostfs-testcases/pytest_tests/helpers/container_access.py

from typing import List, Optional

from frostfs_testlib.shell import Shell
from frostfs_testlib.storage.cluster import Cluster
from frostfs_testlib.storage.dataclasses.acl import EACLOperation
from frostfs_testlib.storage.dataclasses.wallet import WalletInfo

from pytest_tests.helpers.object_access import (
    can_delete_object,
    can_get_head_object,
    can_get_object,
    can_get_range_hash_of_object,
    can_get_range_of_object,
    can_put_object,
    can_search_object,
)


def check_full_access_to_container(
    wallet: WalletInfo,
    cid: str,
    oid: str,
    file_name: str,
    shell: Shell,
    cluster: Cluster,
    bearer: Optional[str] = None,
    xhdr: Optional[dict] = None,
):
    endpoint = cluster.default_rpc_endpoint
    assert can_put_object(wallet, cid, file_name, shell, cluster, bearer, xhdr)
    assert can_get_head_object(wallet, cid, oid, shell, endpoint, bearer, xhdr)
    assert can_get_range_of_object(wallet, cid, oid, shell, endpoint, bearer, xhdr)
    assert can_get_range_hash_of_object(wallet, cid, oid, shell, endpoint, bearer, xhdr)
    assert can_search_object(wallet, cid, shell, endpoint, oid, bearer, xhdr)
    assert can_get_object(wallet, cid, oid, file_name, shell, cluster, bearer, xhdr)
    assert can_delete_object(wallet, cid, oid, shell, endpoint, bearer, xhdr)


def check_no_access_to_container(
    wallet: WalletInfo,
    cid: str,
    oid: str,
    file_name: str,
    shell: Shell,
    cluster: Cluster,
    bearer: Optional[str] = None,
    xhdr: Optional[dict] = None,
):
    endpoint = cluster.default_rpc_endpoint
    assert not can_put_object(wallet, cid, file_name, shell, cluster, bearer, xhdr)
    assert not can_get_head_object(wallet, cid, oid, shell, endpoint, bearer, xhdr)
    assert not can_get_range_of_object(wallet, cid, oid, shell, endpoint, bearer, xhdr)
    assert not can_get_range_hash_of_object(wallet, cid, oid, shell, endpoint, bearer, xhdr)
    assert not can_search_object(wallet, cid, shell, endpoint, oid, bearer, xhdr)
    assert not can_get_object(wallet, cid, oid, file_name, shell, cluster, bearer, xhdr)
    assert not can_delete_object(wallet, cid, oid, shell, endpoint, bearer, xhdr)


def check_custom_access_to_container(
    wallet: WalletInfo,
    cid: str,
    oid: str,
    file_name: str,
    shell: Shell,
    cluster: Cluster,
    deny_operations: Optional[List[EACLOperation]] = None,
    ignore_operations: Optional[List[EACLOperation]] = None,
    bearer: Optional[str] = None,
    xhdr: Optional[dict] = None,
):
    endpoint = cluster.default_rpc_endpoint
    deny_operations = [op.value for op in deny_operations or []]
    ignore_operations = [op.value for op in ignore_operations or []]
    checks: dict = {}
    if EACLOperation.PUT.value not in ignore_operations:
        checks[EACLOperation.PUT.value] = can_put_object(wallet, cid, file_name, shell, cluster, bearer, xhdr)
    if EACLOperation.HEAD.value not in ignore_operations:
        checks[EACLOperation.HEAD.value] = can_get_head_object(wallet, cid, oid, shell, endpoint, bearer, xhdr)
    if EACLOperation.GET_RANGE.value not in ignore_operations:
        checks[EACLOperation.GET_RANGE.value] = can_get_range_of_object(wallet, cid, oid, shell, endpoint, bearer, xhdr)
    if EACLOperation.GET_RANGE_HASH.value not in ignore_operations:
        checks[EACLOperation.GET_RANGE_HASH.value] = can_get_range_hash_of_object(
            wallet, cid, oid, shell, endpoint, bearer, xhdr
        )
    if EACLOperation.SEARCH.value not in ignore_operations:
        checks[EACLOperation.SEARCH.value] = can_search_object(wallet, cid, shell, endpoint, oid, bearer, xhdr)
    if EACLOperation.GET.value not in ignore_operations:
        checks[EACLOperation.GET.value] = can_get_object(wallet, cid, oid, file_name, shell, cluster, bearer, xhdr)
    if EACLOperation.DELETE.value not in ignore_operations:
        checks[EACLOperation.DELETE.value] = can_delete_object(wallet, cid, oid, shell, endpoint, bearer, xhdr)

    failed_checks = [
        f"allowed {action} failed"
        for action, success in checks.items()
        if not success and action not in deny_operations
    ] + [f"denied {action} succeeded" for action, success in checks.items() if success and action in deny_operations]

    assert not failed_checks, ", ".join(failed_checks)


def check_read_only_container(
    wallet: WalletInfo,
    cid: str,
    oid: str,
    file_name: str,
    shell: Shell,
    cluster: Cluster,
    bearer: Optional[str] = None,
    xhdr: Optional[dict] = None,
):
    return check_custom_access_to_container(
        wallet,
        cid,
        oid,
        file_name,
        deny_operations=[EACLOperation.PUT, EACLOperation.DELETE],
        bearer=bearer,
        xhdr=xhdr,
        shell=shell,
        cluster=cluster,
    )