From f2f3d3c8e3814ababf38158a97d94cfad0d386a5 Mon Sep 17 00:00:00 2001 From: Yaroslava Lukoyanova Date: Mon, 5 Jun 2023 12:10:32 +0300 Subject: [PATCH] Add get_data_directory function --- src/frostfs_testlib/hosting/docker_host.py | 4 ++++ src/frostfs_testlib/hosting/interfaces.py | 12 ++++++++++++ .../storage/dataclasses/frostfs_services.py | 3 +++ 3 files changed, 19 insertions(+) diff --git a/src/frostfs_testlib/hosting/docker_host.py b/src/frostfs_testlib/hosting/docker_host.py index 1e32340..5dcac9e 100644 --- a/src/frostfs_testlib/hosting/docker_host.py +++ b/src/frostfs_testlib/hosting/docker_host.py @@ -135,6 +135,10 @@ class DockerHost(Host): timeout=service_attributes.start_timeout, ) + def get_data_directory(self, service_name: str) -> str: + service_attributes = self._get_service_attributes(service_name) + return service_attributes.data_directory_path + def delete_metabase(self, service_name: str) -> None: raise NotImplementedError("Not implemented for docker") diff --git a/src/frostfs_testlib/hosting/interfaces.py b/src/frostfs_testlib/hosting/interfaces.py index 95536c6..8d889da 100644 --- a/src/frostfs_testlib/hosting/interfaces.py +++ b/src/frostfs_testlib/hosting/interfaces.py @@ -112,6 +112,18 @@ class Host(ABC): service_name: Name of the service to restart. """ + + @abstractmethod + def get_data_directory(self, service_name: str) -> str: + """ + Getting path to data directory on node for further usage + (example: list databases pilorama.db) + + Args: + service_name: Name of storage node service. + """ + + @abstractmethod def wait_success_suspend_process(self, process_name: str) -> None: """Search for a service ID by its name and stop the process diff --git a/src/frostfs_testlib/storage/dataclasses/frostfs_services.py b/src/frostfs_testlib/storage/dataclasses/frostfs_services.py index 7bb4c2b..2b52c1f 100644 --- a/src/frostfs_testlib/storage/dataclasses/frostfs_services.py +++ b/src/frostfs_testlib/storage/dataclasses/frostfs_services.py @@ -167,6 +167,9 @@ class StorageNode(NodeBase): def get_un_locode(self): return self._get_attribute(ConfigAttributes.UN_LOCODE) + + def get_data_directory(self) -> str: + return self.host.get_data_directory(self.name) def delete_blobovnicza(self): self.host.delete_blobovnicza(self.name)