frostfs-testlib/src/frostfs_testlib/storage/grpc_operations/interfaces/object.py

from abc import ABC, abstractmethod
from typing import Any, List, Optional

from frostfs_testlib.shell.interfaces import CommandResult
from frostfs_testlib.storage.cluster import Cluster, ClusterNode
from frostfs_testlib.utils import file_utils

from .chunks import ChunksInterface


class ObjectInterface(ABC):
    def __init__(self) -> None:
        self.chunks: ChunksInterface

    @abstractmethod
    def delete(
        self,
        cid: str,
        oid: str,
        endpoint: str,
        bearer: str = "",
        xhdr: Optional[dict] = None,
        session: Optional[str] = None,
        timeout: Optional[str] = None,
    ) -> str:
        pass

    @abstractmethod
    def get(
        self,
        cid: str,
        oid: str,
        endpoint: str,
        bearer: Optional[str] = None,
        write_object: Optional[str] = None,
        xhdr: Optional[dict] = None,
        no_progress: bool = True,
        session: Optional[str] = None,
        timeout: Optional[str] = None,
    ) -> file_utils.TestFile:
        pass

    @abstractmethod
    def get_from_random_node(
        self,
        cid: str,
        oid: str,
        cluster: Cluster,
        bearer: Optional[str] = None,
        write_object: Optional[str] = None,
        xhdr: Optional[dict] = None,
        no_progress: bool = True,
        session: Optional[str] = None,
        timeout: Optional[str] = None,
    ) -> str:
        pass

    @abstractmethod
    def hash(
        self,
        endpoint: str,
        cid: str,
        oid: str,
        address: Optional[str] = None,
        bearer: Optional[str] = None,
        generate_key: Optional[bool] = None,
        range: Optional[str] = None,
        salt: Optional[str] = None,
        ttl: Optional[int] = None,
        session: Optional[str] = None,
        hash_type: Optional[str] = None,
        xhdr: Optional[dict] = None,
        timeout: Optional[str] = None,
    ) -> str:
        pass

    @abstractmethod
    def head(
        self,
        cid: str,
        oid: str,
        endpoint: str,
        bearer: str = "",
        xhdr: Optional[dict] = None,
        json_output: bool = True,
        is_raw: bool = False,
        is_direct: bool = False,
        session: Optional[str] = None,
        timeout: Optional[str] = None,
    ) -> CommandResult | Any:
        pass

    @abstractmethod
    def lock(
        self,
        cid: str,
        oid: str,
        endpoint: str,
        lifetime: Optional[int] = None,
        expire_at: Optional[int] = None,
        address: Optional[str] = None,
        bearer: Optional[str] = None,
        session: Optional[str] = None,
        ttl: Optional[int] = None,
        xhdr: Optional[dict] = None,
        timeout: Optional[str] = None,
    ) -> str:
        pass

    @abstractmethod
    def put(
        self,
        path: str,
        cid: str,
        endpoint: str,
        bearer: Optional[str] = None,
        copies_number: Optional[int] = None,
        attributes: Optional[dict] = None,
        xhdr: Optional[dict] = None,
        expire_at: Optional[int] = None,
        no_progress: bool = True,
        session: Optional[str] = None,
        timeout: Optional[str] = None,
    ) -> str:
        pass

    @abstractmethod
    def patch(
        self,
        cid: str,
        oid: str,
        endpoint: str,
        ranges: Optional[list[str]] = None,
        payloads: Optional[list[str]] = None,
        new_attrs: Optional[str] = None,
        replace_attrs: bool = False,
        bearer: Optional[str] = None,
        xhdr: Optional[dict] = None,
        session: Optional[str] = None,
        timeout: Optional[str] = None,
        trace: bool = False,
    ) -> str:
        pass

    @abstractmethod
    def put_to_random_node(
        self,
        path: str,
        cid: str,
        cluster: Cluster,
        bearer: Optional[str] = None,
        copies_number: Optional[int] = None,
        attributes: Optional[dict] = None,
        xhdr: Optional[dict] = None,
        expire_at: Optional[int] = None,
        no_progress: bool = True,
        session: Optional[str] = None,
        timeout: Optional[str] = None,
    ) -> str:
        pass

    @abstractmethod
    def range(
        self,
        cid: str,
        oid: str,
        range_cut: str,
        endpoint: str,
        bearer: str = "",
        xhdr: Optional[dict] = None,
        session: Optional[str] = None,
        timeout: Optional[str] = None,
    ) -> tuple[file_utils.TestFile, bytes]:
        pass

    @abstractmethod
    def search(
        self,
        cid: str,
        endpoint: str,
        bearer: str = "",
        oid: Optional[str] = None,
        filters: Optional[dict] = None,
        expected_objects_list: Optional[list] = None,
        xhdr: Optional[dict] = None,
        session: Optional[str] = None,
        phy: bool = False,
        root: bool = False,
        timeout: Optional[str] = None,
        address: Optional[str] = None,
        generate_key: Optional[bool] = None,
        ttl: Optional[int] = None,
    ) -> List:
        pass

    @abstractmethod
    def nodes(
        self,
        cluster: Cluster,
        cid: str,
        oid: str,
        alive_node: ClusterNode,
        bearer: str = "",
        xhdr: Optional[dict] = None,
        is_direct: bool = False,
        verify_presence_all: bool = False,
        timeout: Optional[str] = None,
    ) -> List[ClusterNode]:
        pass

    @abstractmethod
    def parts(
        self,
        cid: str,
        oid: str,
        alive_node: ClusterNode,
        bearer: str = "",
        xhdr: Optional[dict] = None,
        is_direct: bool = False,
        verify_presence_all: bool = False,
        timeout: Optional[str] = None,
    ) -> List[str]:
        pass