diff --git a/.gitignore b/.gitignore index 8a7034d..c3d58f7 100644 --- a/.gitignore +++ b/.gitignore @@ -7,4 +7,5 @@ # ignore build artifacts /dist +/build *.egg-info diff --git a/src/neofs_testlib/hosting/docker_host.py b/src/neofs_testlib/hosting/docker_host.py index ae97a32..ed3604c 100644 --- a/src/neofs_testlib/hosting/docker_host.py +++ b/src/neofs_testlib/hosting/docker_host.py @@ -10,7 +10,7 @@ import docker from requests import HTTPError from neofs_testlib.hosting.config import ParsedAttributes -from neofs_testlib.hosting.interfaces import Host +from neofs_testlib.hosting.interfaces import DiskInfo, Host from neofs_testlib.shell import LocalShell, Shell, SSHShell from neofs_testlib.shell.command_inspectors import SudoInspector @@ -127,6 +127,15 @@ class DockerHost(Host): cmd = f"rm -rf {volume_path}/meta*" if cache_only else f"rm -rf {volume_path}/*" shell.exec(cmd) + def attach_disk(self, device: str, disk_info: DiskInfo) -> None: + raise NotImplementedError("Not supported for docker") + + def detach_disk(self, device: str) -> DiskInfo: + raise NotImplementedError("Not supported for docker") + + def is_disk_attached(self, device: str, disk_info: DiskInfo) -> bool: + raise NotImplementedError("Not supported for docker") + def dump_logs( self, directory_path: str, diff --git a/src/neofs_testlib/hosting/interfaces.py b/src/neofs_testlib/hosting/interfaces.py index 50eda0d..b90eb3d 100644 --- a/src/neofs_testlib/hosting/interfaces.py +++ b/src/neofs_testlib/hosting/interfaces.py @@ -1,11 +1,15 @@ from abc import ABC, abstractmethod from datetime import datetime -from typing import Optional +from typing import Any, Optional from neofs_testlib.hosting.config import CLIConfig, HostConfig, ServiceConfig from neofs_testlib.shell.interfaces import Shell +class DiskInfo(dict): + """Dict wrapper for disk_info for disk management commands.""" + + class Host(ABC): """Interface of a host machine where neoFS services are running. @@ -109,6 +113,40 @@ class Host(ABC): cache_only: To delete cache only. """ + @abstractmethod + def detach_disk(self, device: str) -> DiskInfo: + """Detaches disk device to simulate disk offline/failover scenario. + + Args: + device: Device name to detach + + Returns: + internal service disk info related to host plugin (i.e. volume id for cloud devices), + which may be used to identify or re-attach existing volume back + """ + + @abstractmethod + def attach_disk(self, device: str, disk_info: DiskInfo) -> None: + """Attaches disk device back. + + Args: + device: Device name to attach + service_info: any info required for host plugin to identify/attach disk + """ + + @abstractmethod + def is_disk_attached(self, device: str, disk_info: DiskInfo) -> bool: + """Checks if disk device is attached. + + Args: + device: Device name to check + service_info: any info required for host plugin to identify disk + + Returns: + True if attached + False if detached + """ + @abstractmethod def dump_logs( self,