Add method to interfaces
Signed-off-by: Dmitriy Zayakin <d.zayakin@yadro.com>
This commit is contained in:
parent
e1f3444e92
commit
0c3bb20af5
3 changed files with 52 additions and 2 deletions
|
@ -17,6 +17,7 @@ from frostfs_testlib.storage.dataclasses.frostfs_services import (
|
|||
StorageNode,
|
||||
)
|
||||
from frostfs_testlib.storage.dataclasses.node_base import NodeBase, ServiceClass
|
||||
from frostfs_testlib.storage.dataclasses.storage_object_info import Interfaces
|
||||
from frostfs_testlib.storage.service_registry import ServiceRegistry
|
||||
|
||||
reporter = get_reporter()
|
||||
|
@ -121,6 +122,40 @@ class ClusterNode:
|
|||
config.attributes[ConfigAttributes.SERVICE_NAME] for config in self.host.config.services
|
||||
]
|
||||
|
||||
def get_all_interfaces(self) -> dict[str, str]:
|
||||
return self.host.config.interfaces
|
||||
|
||||
def get_interface(self, interface: Interfaces) -> str:
|
||||
return self.host.config.interfaces[interface.value]
|
||||
|
||||
def get_data_interfaces(self) -> list[str]:
|
||||
return [
|
||||
ip_address
|
||||
for name_interface, ip_address in self.host.config.interfaces.items()
|
||||
if "data" in name_interface
|
||||
]
|
||||
|
||||
def get_data_interface(self, search_interface: str) -> list[str]:
|
||||
return [
|
||||
self.host.config.interfaces[interface]
|
||||
for interface in self.host.config.interfaces.keys()
|
||||
if search_interface == interface
|
||||
]
|
||||
|
||||
def get_internal_interfaces(self) -> list[str]:
|
||||
return [
|
||||
ip_address
|
||||
for name_interface, ip_address in self.host.config.interfaces.items()
|
||||
if "internal" in name_interface
|
||||
]
|
||||
|
||||
def get_internal_interface(self, search_internal: str) -> list[str]:
|
||||
return [
|
||||
self.host.config.interfaces[interface]
|
||||
for interface in self.host.config.interfaces.keys()
|
||||
if search_internal == interface
|
||||
]
|
||||
|
||||
|
||||
class Cluster:
|
||||
"""
|
||||
|
|
|
@ -110,6 +110,10 @@ class MorphChain(NodeBase):
|
|||
def label(self) -> str:
|
||||
return f"{self.name}: {self.get_endpoint()}"
|
||||
|
||||
def get_http_endpoint(self) -> str:
|
||||
return self._get_attribute("http_endpoint")
|
||||
|
||||
|
||||
class StorageNode(NodeBase):
|
||||
"""
|
||||
Class represents storage node in a storage cluster
|
||||
|
@ -149,10 +153,10 @@ class StorageNode(NodeBase):
|
|||
|
||||
def get_data_directory(self) -> str:
|
||||
return self.host.get_data_directory(self.name)
|
||||
|
||||
|
||||
def get_http_hostname(self) -> str:
|
||||
return self._get_attribute(ConfigAttributes.HTTP_HOSTNAME)
|
||||
|
||||
|
||||
def get_s3_hostname(self) -> str:
|
||||
return self._get_attribute(ConfigAttributes.S3_HOSTNAME)
|
||||
|
||||
|
|
|
@ -1,6 +1,9 @@
|
|||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import Optional
|
||||
|
||||
from frostfs_testlib.testing.readable import HumanReadableEnum
|
||||
|
||||
|
||||
@dataclass
|
||||
class ObjectRef:
|
||||
|
@ -42,3 +45,11 @@ class NodeNetmapInfo:
|
|||
sub_div_code: int = None
|
||||
un_locode: str = None
|
||||
role: str = None
|
||||
|
||||
|
||||
class Interfaces(HumanReadableEnum):
|
||||
DATA_O: str = "data0"
|
||||
DATA_1: str = "data1"
|
||||
MGMT: str = "mgmt"
|
||||
INTERNAL_0: str = "internal0"
|
||||
INTERNAL_1: str = "internal1"
|
||||
|
|
Loading…
Reference in a new issue