test-api: reformat api files

Signed-off-by: Kamil Gierszewski <kamil.gierszewski@huawei.com>
This commit is contained in:
Kamil Gierszewski
2024-08-09 10:52:00 +02:00
parent 077820f1c0
commit d40e2a519d
11 changed files with 962 additions and 837 deletions

View File

@@ -5,6 +5,7 @@
#
from api.cas.casadm_parser import *
from api.cas.core import Core
from api.cas.dmesg import get_metadata_size_on_device
from api.cas.statistics import CacheStats, IoClassStats
from test_utils.os_utils import *
@@ -12,7 +13,7 @@ from test_utils.output import Output
class Cache:
def __init__(self, device: Device, cache_id: int = None):
def __init__(self, device: Device, cache_id: int = None) -> None:
self.cache_device = device
self.cache_id = cache_id if cache_id else self.__get_cache_id()
self.__cache_line_size = None
@@ -32,17 +33,17 @@ class Cache:
def __get_cache_device_path(self) -> str:
return self.cache_device.path if self.cache_device is not None else "-"
def get_core_devices(self):
def get_core_devices(self) -> list:
return get_cores(self.cache_id)
def get_cache_line_size(self):
def get_cache_line_size(self) -> CacheLineSize:
if self.__cache_line_size is None:
stats = self.get_statistics()
stats_line_size = stats.config_stats.cache_line_size
self.__cache_line_size = CacheLineSize(stats_line_size)
return self.__cache_line_size
def get_cleaning_policy(self):
def get_cleaning_policy(self) -> CleaningPolicy:
stats = self.get_statistics()
cp = stats.config_stats.cleaning_policy
return CleaningPolicy[cp]
@@ -58,7 +59,7 @@ class Cache:
def get_occupancy(self):
return self.get_statistics().usage_stats.occupancy
def get_status(self):
def get_status(self) -> CacheStatus:
status = (
self.get_statistics(stat_filter=[StatsFilter.conf])
.config_stats.status.replace(" ", "_")
@@ -67,25 +68,25 @@ class Cache:
return CacheStatus[status]
@property
def size(self):
def size(self) -> Size:
return self.get_statistics().config_stats.cache_size
def get_cache_mode(self):
def get_cache_mode(self) -> CacheMode:
return CacheMode[self.get_statistics().config_stats.write_policy.upper()]
def get_dirty_blocks(self):
def get_dirty_blocks(self) -> Size:
return self.get_statistics().usage_stats.dirty
def get_dirty_for(self):
def get_dirty_for(self) -> timedelta:
return self.get_statistics().config_stats.dirty_for
def get_clean_blocks(self):
def get_clean_blocks(self) -> Size:
return self.get_statistics().usage_stats.clean
def get_flush_parameters_alru(self):
def get_flush_parameters_alru(self) -> FlushParametersAlru:
return get_flush_parameters_alru(self.cache_id)
def get_flush_parameters_acp(self):
def get_flush_parameters_acp(self) -> FlushParametersAcp:
return get_flush_parameters_acp(self.cache_id)
# Casadm methods:
@@ -106,7 +107,7 @@ class Cache:
io_class_id: int,
stat_filter: List[StatsFilter] = None,
percentage_val: bool = False,
):
) -> IoClassStats:
return IoClassStats(
cache_id=self.cache_id,
filter=stat_filter,
@@ -115,39 +116,40 @@ class Cache:
)
def flush_cache(self) -> Output:
cmd_output = casadm.flush_cache(cache_id=self.cache_id)
output = casadm.flush_cache(cache_id=self.cache_id)
sync()
return cmd_output
return output
def purge_cache(self):
casadm.purge_cache(cache_id=self.cache_id)
def purge_cache(self) -> Output:
output = casadm.purge_cache(cache_id=self.cache_id)
sync()
return output
def stop(self, no_data_flush: bool = False):
def stop(self, no_data_flush: bool = False) -> Output:
return casadm.stop_cache(self.cache_id, no_data_flush)
def add_core(self, core_dev, core_id: int = None):
def add_core(self, core_dev, core_id: int = None) -> Core:
return casadm.add_core(self, core_dev, core_id)
def remove_core(self, core_id: int, force: bool = False):
def remove_core(self, core_id: int, force: bool = False) -> Output:
return casadm.remove_core(self.cache_id, core_id, force)
def remove_inactive_core(self, core_id: int, force: bool = False):
def remove_inactive_core(self, core_id: int, force: bool = False) -> Output:
return casadm.remove_inactive(self.cache_id, core_id, force)
def reset_counters(self):
def reset_counters(self) -> Output:
return casadm.reset_counters(self.cache_id)
def set_cache_mode(self, cache_mode: CacheMode, flush=None):
def set_cache_mode(self, cache_mode: CacheMode, flush=None) -> Output:
return casadm.set_cache_mode(cache_mode, self.cache_id, flush)
def load_io_class(self, file_path: str):
def load_io_class(self, file_path: str) -> Output:
return casadm.load_io_classes(self.cache_id, file_path)
def list_io_classes(self):
def list_io_classes(self) -> list:
return get_io_class_list(self.cache_id)
def set_seq_cutoff_parameters(self, seq_cutoff_param: SeqCutOffParameters):
def set_seq_cutoff_parameters(self, seq_cutoff_param: SeqCutOffParameters) -> Output:
return casadm.set_param_cutoff(
self.cache_id,
threshold=seq_cutoff_param.threshold,
@@ -155,16 +157,16 @@ class Cache:
promotion_count=seq_cutoff_param.promotion_count,
)
def set_seq_cutoff_threshold(self, threshold: Size):
def set_seq_cutoff_threshold(self, threshold: Size) -> Output:
return casadm.set_param_cutoff(self.cache_id, threshold=threshold, policy=None)
def set_seq_cutoff_policy(self, policy: SeqCutOffPolicy):
def set_seq_cutoff_policy(self, policy: SeqCutOffPolicy) -> Output:
return casadm.set_param_cutoff(self.cache_id, threshold=None, policy=policy)
def set_cleaning_policy(self, cleaning_policy: CleaningPolicy):
def set_cleaning_policy(self, cleaning_policy: CleaningPolicy) -> Output:
return casadm.set_param_cleaning(self.cache_id, cleaning_policy)
def set_params_acp(self, acp_params: FlushParametersAcp):
def set_params_acp(self, acp_params: FlushParametersAcp) -> Output:
return casadm.set_param_cleaning_acp(
self.cache_id,
(
@@ -175,7 +177,7 @@ class Cache:
int(acp_params.flush_max_buffers) if acp_params.flush_max_buffers else None,
)
def set_params_alru(self, alru_params: FlushParametersAlru):
def set_params_alru(self, alru_params: FlushParametersAlru) -> Output:
return casadm.set_param_cleaning_alru(
self.cache_id,
(
@@ -196,17 +198,17 @@ class Cache:
),
)
def get_cache_config(self):
def get_cache_config(self) -> CacheConfig:
return CacheConfig(
self.get_cache_line_size(),
self.get_cache_mode(),
self.get_cleaning_policy(),
)
def standby_detach(self, shortcut: bool = False):
def standby_detach(self, shortcut: bool = False) -> Output:
return casadm.standby_detach_cache(cache_id=self.cache_id, shortcut=shortcut)
def standby_activate(self, device, shortcut: bool = False):
def standby_activate(self, device, shortcut: bool = False) -> Output:
return casadm.standby_activate_cache(
cache_id=self.cache_id, cache_dev=device, shortcut=shortcut
)