test-api: refactor to fix circular dependencies

Signed-off-by: Kamil Gierszewski <[email protected]>

test/functional/api/cas/cache_config.py

@@ -125,6 +125,17 @@ class CacheStatus(Enum):
     incomplete = "incomplete"
     standby = "standby"
     standby_detached = "standby detached"
+    detached = "detached"
+
+    def __str__(self):
+        return self.value
+
+
+class CoreStatus(Enum):
+    empty = "empty"
+    active = "active"
+    inactive = "inactive"
+    detached = "detached"
 
     def __str__(self):
         return self.value

test/functional/api/cas/casadm_parser.py

@@ -5,7 +5,6 @@
 #
 
 import csv
-import io
 import json
 
 from datetime import timedelta, datetime
@@ -45,8 +44,8 @@ def get_caches() -> list:
     for cache in caches_dict.values():
         caches_list.append(
             Cache(
+                cache_id=(cache["id"]),
                 device=(Device(cache["device_path"]) if cache["device_path"] != "-" else None),
-                cache_id=cache["id"],
             )
         )
 
@@ -59,7 +58,7 @@ def get_cores(cache_id: int) -> list:
     cores_dict = get_cas_devices_dict()["cores"].values()
 
     def is_active(core):
-        return CoreStatus[core["status"].lower()] == CoreStatus.active
+        return core["status"] == CoreStatus.active
 
     return [
         Core(core["device_path"], core["cache_id"])
@@ -74,7 +73,7 @@ def get_inactive_cores(cache_id: int) -> list:
     cores_dict = get_cas_devices_dict()["cores"].values()
 
     def is_inactive(core):
-        return CoreStatus[core["status"].lower()] == CoreStatus.inactive
+        return core["status"] == CoreStatus.inactive
 
     return [
         Core(core["device_path"], core["cache_id"])
@@ -89,7 +88,7 @@ def get_detached_cores(cache_id: int) -> list:
     cores_dict = get_cas_devices_dict()["cores"].values()
 
     def is_detached(core):
-        return CoreStatus[core["status"].lower()] == CoreStatus.detached
+        return core["status"] == CoreStatus.detached
 
     return [
         Core(core["device_path"], core["cache_id"])
@@ -110,15 +109,17 @@ def get_cas_devices_dict() -> dict:
             params = [
                 ("id", cache_id),
                 ("device_path", device["disk"]),
-                ("status", device["status"]),
+                ("status", CacheStatus(device["status"].lower())),
             ]
             devices["caches"][cache_id] = dict([(key, value) for key, value in params])
 
         elif device["type"] == "core":
             params = [
                 ("cache_id", cache_id),
+                ("core_id", (int(device["id"]) if device["id"] != "-" else device["id"])),
                 ("device_path", device["disk"]),
-                ("status", device["status"]),
+                ("status", CoreStatus(device["status"].lower())),
+                ("exp_obj", device["device"]),
             ]
             if core_pool:
                 params.append(("core_pool", device))
@@ -201,21 +202,6 @@ def get_flush_parameters_acp(cache_id: int):
     return flush_parameters
 
 
-def get_seq_cut_off_parameters(cache_id: int, core_id: int):
-    casadm_output = casadm.get_param_cutoff(
-        cache_id, core_id, casadm.OutputFormat.csv
-    ).stdout.splitlines()
-    seq_cut_off_params = SeqCutOffParameters()
-    for line in casadm_output:
-        if "Sequential cutoff threshold" in line:
-            seq_cut_off_params.threshold = Size(int(line.split(",")[1]), Unit.KibiByte)
-        if "Sequential cutoff policy" in line:
-            seq_cut_off_params.policy = SeqCutOffPolicy.from_name(line.split(",")[1])
-        if "Sequential cutoff promotion request count threshold" in line:
-            seq_cut_off_params.promotion_count = int(line.split(",")[1])
-    return seq_cut_off_params
-
-
 def get_casadm_version():
     casadm_output = casadm.print_version(OutputFormat.csv).stdout.split("\n")
     version_str = casadm_output[1].split(",")[-1]
@@ -231,21 +217,3 @@ def get_io_class_list(cache_id: int) -> list:
         ioclass = IoClass(int(values[0]), values[1], int(values[2]), values[3])
         ret.append(ioclass)
     return ret
-
-
-def get_core_info_for_cache_by_path(core_disk_path: str, target_cache_id: int) -> dict | None:
-    output = casadm.list_caches(OutputFormat.csv, by_id_path=True)
-    reader = csv.DictReader(io.StringIO(output.stdout))
-    cache_id = -1
-    for row in reader:
-        if row["type"] == "cache":
-            cache_id = int(row["id"])
-        if row["type"] == "core" and row["disk"] == core_disk_path and target_cache_id == cache_id:
-            return {
-                "core_id": row["id"],
-                "core_device": row["disk"],
-                "status": row["status"],
-                "exp_obj": row["device"],
-            }
-
-    return None

test/functional/api/cas/core.py

@@ -6,12 +6,11 @@
 
 from datetime import timedelta
 from typing import List
-from enum import Enum
 
 from api.cas import casadm
-from api.cas.cache_config import SeqCutOffParameters, SeqCutOffPolicy
+from api.cas.cache_config import SeqCutOffParameters, SeqCutOffPolicy, CoreStatus
 from api.cas.casadm_params import StatsFilter
-from api.cas.casadm_parser import get_seq_cut_off_parameters, get_core_info_for_cache_by_path
+from api.cas.casadm_parser import get_cas_devices_dict
 from api.cas.statistics import CoreStats, CoreIoClassStats
 from core.test_run_utils import TestRun
 from storage_devices.device import Device
@@ -20,13 +19,6 @@
 from test_utils.size import Unit, Size
 
 
-class CoreStatus(Enum):
-    empty = 0
-    active = 1
-    inactive = 2
-    detached = 3
-
-
 SEQ_CUTOFF_THRESHOLD_MAX = Size(4194181, Unit.KibiByte)
 SEQ_CUT_OFF_THRESHOLD_DEFAULT = Size(1, Unit.MebiByte)
 
@@ -42,12 +34,14 @@ def __init__(self, core_device: str, cache_id: int):
             self.core_id = int(core_info["core_id"])
         if core_info["exp_obj"] != "-":
             Device.__init__(self, core_info["exp_obj"])
+        self.status = None
         self.partitions = []
         self.block_size = None
 
-    def __get_core_info(self):
-        return get_core_info_for_cache_by_path(core_disk_path=self.core_device.path,
-                                               target_cache_id=self.cache_id)
+    def __get_core_info(self) -> dict | None:
+        core_dicts = get_cas_devices_dict()["cores"].values()
+        return next(iter([core for core in core_dicts if core["cache_id"] == self.cache_id and
+                          core["device_path"] == self.core_device.path]))
 
     def create_filesystem(self, fs_type: disk_utils.Filesystem, force=True, blocksize=None):
         super().create_filesystem(fs_type, force, blocksize)
@@ -77,17 +71,32 @@ def get_statistics(
             percentage_val=percentage_val,
         )
 
-    def get_status(self):
-        return CoreStatus[self.__get_core_info()["status"].lower()]
+    def get_status(self) -> CoreStatus:
+        self.status = self.__get_core_info()["status"]
+        return self.status
+
+    def __get_seq_cut_off_parameters(self):
+        casadm_output = casadm.get_param_cutoff(
+            self.cache_id, self.core_id, casadm.OutputFormat.csv
+        ).stdout.splitlines()
+        seq_cut_off_params = SeqCutOffParameters()
+        for line in casadm_output:
+            if "Sequential cutoff threshold" in line:
+                seq_cut_off_params.threshold = Size(int(line.split(",")[1]), Unit.KibiByte)
+            if "Sequential cutoff policy" in line:
+                seq_cut_off_params.policy = SeqCutOffPolicy.from_name(line.split(",")[1])
+            if "Sequential cutoff promotion request count threshold" in line:
+                seq_cut_off_params.promotion_count = int(line.split(",")[1])
+        return seq_cut_off_params
 
     def get_seq_cut_off_parameters(self):
-        return get_seq_cut_off_parameters(self.cache_id, self.core_id)
+        return self.__get_seq_cut_off_parameters(self.cache_id, self.core_id)
 
     def get_seq_cut_off_policy(self):
-        return get_seq_cut_off_parameters(self.cache_id, self.core_id).policy
+        return self.__get_seq_cut_off_parameters(self.cache_id, self.core_id).policy
 
     def get_seq_cut_off_threshold(self):
-        return get_seq_cut_off_parameters(self.cache_id, self.core_id).threshold
+        return self.__get_seq_cut_off_parameters(self.cache_id, self.core_id).threshold
 
     def get_dirty_blocks(self):
         return self.get_statistics().usage_stats.dirty