[#372] Added decorator wait until stabilization metric values
All checks were successful
DCO action / DCO (pull_request) Successful in 22s

Signed-off-by: Ilyas Niyazov <i.niyazov@yadro.com>
This commit is contained in:
Ilyas Niyazov 2025-04-09 16:15:46 +03:00
parent d38808a1f5
commit 733a69529e

View file

@ -1,3 +1,9 @@
import time
from functools import wraps
from typing import Callable
import pytest
from frostfs_testlib.hosting import Host from frostfs_testlib.hosting import Host
from frostfs_testlib.shell.interfaces import CommandResult from frostfs_testlib.shell.interfaces import CommandResult
@ -7,11 +13,11 @@ class Metrics:
self.storage = StorageMetrics(host, metrics_endpoint) self.storage = StorageMetrics(host, metrics_endpoint)
class StorageMetrics: class StorageMetrics:
""" """
Class represents storage metrics in a cluster Class represents storage metrics in a cluster
""" """
def __init__(self, host: Host, metrics_endpoint: str) -> None: def __init__(self, host: Host, metrics_endpoint: str) -> None:
self.host = host self.host = host
self.metrics_endpoint = metrics_endpoint self.metrics_endpoint = metrics_endpoint
@ -34,3 +40,41 @@ class StorageMetrics:
shell = self.host.get_shell() shell = self.host.get_shell()
result = shell.exec(f"curl -s {self.metrics_endpoint}") result = shell.exec(f"curl -s {self.metrics_endpoint}")
return result return result
def wait_until_metric_result_is_stable(
relative_deviation: float = None, absolute_deviation: int = None, max_attempts: int = 10, sleep_interval: int = 30
):
"""
A decorator function that repeatedly calls the decorated function until its result stabilizes
within a specified relative tolerance or until the maximum number of attempts is reached.
This decorator is useful for scenarios where a function returns a metric or value that may fluctuate
over time, and you want to ensure that the result has stabilized before proceeding.
"""
def decorator(func: Callable):
@wraps(func)
def wrapper(*args, **kwargs):
last_result = None
for _ in range(max_attempts):
# first function call
first_result = func(*args, **kwargs)
# waiting before the second call
time.sleep(sleep_interval)
# second function call
last_result = func(*args, **kwargs)
# checking value stability
if first_result == pytest.approx(last_result, rel=relative_deviation, abs=absolute_deviation):
return last_result
# if stability is not achieved, return the last value
if last_result is not None:
return last_result
return wrapper
return decorator