forked from TrueCloudLab/frostfs-testlib
46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
import time
|
|
from typing import Optional
|
|
|
|
from frostfs_testlib.reporter import get_reporter
|
|
from frostfs_testlib.resources.common import MORPH_BLOCK_TIME
|
|
from frostfs_testlib.shell import Shell
|
|
from frostfs_testlib.steps import epoch
|
|
from frostfs_testlib.storage.cluster import Cluster
|
|
from frostfs_testlib.storage.dataclasses.frostfs_services import StorageNode
|
|
from frostfs_testlib.utils import datetime_utils
|
|
|
|
reporter = get_reporter()
|
|
|
|
|
|
# To skip adding every mandatory singleton dependency to EACH test function
|
|
class ClusterTestBase:
|
|
shell: Shell
|
|
cluster: Cluster
|
|
|
|
@reporter.step_deco("Tick {epochs_to_tick} epochs, wait {wait_block} block")
|
|
def tick_epochs(
|
|
self,
|
|
epochs_to_tick: int,
|
|
alive_node: Optional[StorageNode] = None,
|
|
wait_block: int = None,
|
|
):
|
|
for _ in range(epochs_to_tick):
|
|
self.tick_epoch(alive_node, wait_block)
|
|
|
|
def tick_epoch(
|
|
self,
|
|
alive_node: Optional[StorageNode] = None,
|
|
wait_block: int = None,
|
|
):
|
|
epoch.tick_epoch(self.shell, self.cluster, alive_node=alive_node)
|
|
if wait_block:
|
|
time.sleep(datetime_utils.parse_time(MORPH_BLOCK_TIME) * wait_block)
|
|
|
|
def wait_for_epochs_align(self):
|
|
epoch.wait_for_epochs_align(self.shell, self.cluster)
|
|
|
|
def get_epoch(self):
|
|
return epoch.get_epoch(self.shell, self.cluster)
|
|
|
|
def ensure_fresh_epoch(self):
|
|
return epoch.ensure_fresh_epoch(self.shell, self.cluster)
|