Update tests

test/functional/api/cas/cache.py

@@ -13,21 +13,28 @@
 
 
 class Cache:
-    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
-
-    def __get_cache_id(self) -> int:
-        device_path = self.__get_cache_device_path()
-
+    def __init__(
+        self, cache_id: int, device: Device = None, cache_line_size: CacheLineSize = None
+    ) -> None:
+        self.cache_id = cache_id
+        self.cache_device = device if device else self.__get_cache_device()
+        self.status = None
+        self.cache_line_size = cache_line_size
+        self.write_policy = None
+
+    def __get_cache_device(self) -> Device | None:
         caches_dict = get_cas_devices_dict()["caches"]
+        cache = next(
+            iter([cache for cache in caches_dict.values() if cache["id"] == self.cache_id])
+        )
+
+        if not cache:
+            return None
 
-        for cache in caches_dict.values():
-            if cache["device_path"] == device_path:
-                return int(cache["id"])
+        if cache["device_path"] is "-":
+            return None
 
-        raise Exception(f"There is no cache started on {device_path}")
+        return Device(path=cache["device_path"])
 
     def __get_cache_device_path(self) -> str:
         return self.cache_device.path if self.cache_device is not None else "-"
@@ -36,11 +43,11 @@ def get_core_devices(self) -> list:
         return get_cores(self.cache_id)
 
     def get_cache_line_size(self) -> CacheLineSize:
-        if self.__cache_line_size is None:
+        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
+            self.cache_line_size = CacheLineSize(stats_line_size)
+        return self.cache_line_size
 
     def get_cleaning_policy(self) -> CleaningPolicy:
         stats = self.get_statistics()
@@ -195,7 +202,7 @@ def set_params_nhit(self, promotion_params_nhit: PromotionParametersNhit) -> Out
         return casadm.set_param_promotion_nhit(
             self.cache_id,
             threshold=promotion_params_nhit.threshold,
-            trigger=promotion_params_nhit.trigger
+            trigger=promotion_params_nhit.trigger,
         )
 
     def get_cache_config(self) -> CacheConfig:
@@ -208,7 +215,7 @@ def get_cache_config(self) -> CacheConfig:
     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) -> Output:
+    def standby_activate(self, device: Device, shortcut: bool = False) -> Output:
         return casadm.standby_activate_cache(
             cache_id=self.cache_id, cache_dev=device, shortcut=shortcut
         )

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.py

@@ -48,6 +48,7 @@ def start_cache(
     )
     _cache_id = str(cache_id) if cache_id is not None else None
     _cache_mode = cache_mode.name.lower() if cache_mode else None
+
     output = TestRun.executor.run(
         start_cmd(
             cache_dev=cache_dev.path,
@@ -59,16 +60,43 @@ def start_cache(
             shortcut=shortcut,
         )
     )
+
     if output.exit_code != 0:
         raise CmdException("Failed to start cache.", output)
-    return Cache(cache_dev)
+
+    if not _cache_id:
+        from api.cas.casadm_parser import get_caches
+
+        cache_list = get_caches()
+        # compare ids of old and new caches, returning the only one created now.
+        # This will be needed in case cache_id not present in cli command
+
+        new_cache = next(
+            iter([cache for cache in cache_list if cache.cache_device.path == cache_dev.path])
+        )
+        _cache_id = new_cache.cache_id
+
+    cache = Cache(cache_id=int(_cache_id), device=cache_dev, cache_line_size=_cache_line_size)
+    TestRun.dut.cache_list.append(cache)
+    return cache
 
 
 def load_cache(device: Device, shortcut: bool = False) -> Cache:
+    from api.cas.casadm_parser import get_caches
+
+    caches_before_load = get_caches()
     output = TestRun.executor.run(load_cmd(cache_dev=device.path, shortcut=shortcut))
+
     if output.exit_code != 0:
         raise CmdException("Failed to load cache.", output)
-    return Cache(device)
+
+    caches_after_load = get_caches()
+    new_cache = next(
+        iter([cache for cache in caches_after_load if cache not in caches_before_load])
+    )
+    cache = Cache(cache_id=new_cache.cache_id, device=new_cache.cache_device)
+    TestRun.dut.cache_list.append(cache)
+    return cache
 
 
 def attach_cache(cache_id: int, device: Device, force: bool, shortcut: bool = False) -> Output:
@@ -84,17 +112,29 @@ def attach_cache(cache_id: int, device: Device, force: bool, shortcut: bool = Fa
 
 def detach_cache(cache_id: int, shortcut: bool = False) -> Output:
     output = TestRun.executor.run(detach_cache_cmd(cache_id=str(cache_id), shortcut=shortcut))
+
     if output.exit_code != 0:
         raise CmdException("Failed to detach cache.", output)
+
+    detached_cache = next(cache for cache in TestRun.dut.cache_list if cache.cache_id == cache_id)
+    detached_cache.cache_device = None
     return output
 
 
 def stop_cache(cache_id: int, no_data_flush: bool = False, shortcut: bool = False) -> Output:
     output = TestRun.executor.run(
         stop_cmd(cache_id=str(cache_id), no_data_flush=no_data_flush, shortcut=shortcut)
     )
+
     if output.exit_code != 0:
         raise CmdException("Failed to stop cache.", output)
+
+    TestRun.dut.cache_list = [
+        cache for cache in TestRun.dut.cache_list if cache.cache_id != cache_id
+    ]
+
+    TestRun.dut.core_list = [core for core in TestRun.dut.core_list if core.cache_id == cache_id]
+
     return output
 
 
@@ -192,7 +232,7 @@ def set_param_promotion(cache_id: int, policy: PromotionPolicy, shortcut: bool =
 
 
 def set_param_promotion_nhit(
-        cache_id: int, threshold: int = None, trigger: int = None, shortcut: bool = False
+    cache_id: int, threshold: int = None, trigger: int = None, shortcut: bool = False
 ) -> Output:
     _threshold = str(threshold) if threshold is not None else None
     _trigger = str(trigger) if trigger is not None else None
@@ -267,7 +307,7 @@ def get_param_cleaning_acp(
 
 
 def get_param_promotion(
-        cache_id: int, output_format: OutputFormat = None, shortcut: bool = False
+    cache_id: int, output_format: OutputFormat = None, shortcut: bool = False
 ) -> Output:
     _output_format = output_format.name if output_format else None
     output = TestRun.executor.run(
@@ -281,7 +321,7 @@ def get_param_promotion(
 
 
 def get_param_promotion_nhit(
-        cache_id: int, output_format: OutputFormat = None, shortcut: bool = False
+    cache_id: int, output_format: OutputFormat = None, shortcut: bool = False
 ) -> Output:
     _output_format = output_format.name if output_format else None
     output = TestRun.executor.run(
@@ -325,7 +365,11 @@ def add_core(cache: Cache, core_dev: Device, core_id: int = None, shortcut: bool
     )
     if output.exit_code != 0:
         raise CmdException("Failed to add core.", output)
-    return Core(core_dev.path, cache.cache_id)
+
+    core = Core(core_dev.path, cache.cache_id)
+    TestRun.dut.core_list.append(core)
+
+    return core
 
 
 def remove_core(cache_id: int, core_id: int, force: bool = False, shortcut: bool = False) -> Output:
@@ -336,6 +380,17 @@ def remove_core(cache_id: int, core_id: int, force: bool = False, shortcut: bool
     )
     if output.exit_code != 0:
         raise CmdException("Failed to remove core.", output)
+
+    core_to_remove = next(
+        iter(
+            [
+                core
+                for core in TestRun.dut.core_list
+                if core.cache_id == cache_id and core.core_id == core_id
+            ]
+        )
+    )
+    TestRun.dut.core_list.remove(core_to_remove)
     return output
 
 
@@ -485,22 +540,45 @@ def standby_init(
             shortcut=shortcut,
         )
     )
+
     if output.exit_code != 0:
         raise CmdException("Failed to init standby cache.", output)
-    return Cache(cache_dev)
+    return Cache(cache_id=cache_id, device=cache_dev)
 
 
 def standby_load(cache_dev: Device, shortcut: bool = False) -> Cache:
+    from api.cas.casadm_parser import get_caches
+
+    caches_before_load = get_caches()
     output = TestRun.executor.run(standby_load_cmd(cache_dev=cache_dev.path, shortcut=shortcut))
+
     if output.exit_code != 0:
-        raise CmdException("Failed to load standby cache.", output)
-    return Cache(cache_dev)
+        raise CmdException("Failed to load cache.", output)
+    caches_after_load = get_caches()
+    # compare ids of old and new caches, returning the only one created now
+    new_cache = next(
+        iter(
+            [
+                cache
+                for cache in caches_after_load
+                if cache.cache_id not in [cache.cache_id for cache in caches_before_load]
+            ]
+        )
+    )
+    cache = Cache(cache_id=new_cache.cache_id, device=new_cache.cache_device)
+    TestRun.dut.cache_list.append(cache)
+
+    return cache
 
 
 def standby_detach_cache(cache_id: int, shortcut: bool = False) -> Output:
     output = TestRun.executor.run(standby_detach_cmd(cache_id=str(cache_id), shortcut=shortcut))
     if output.exit_code != 0:
         raise CmdException("Failed to detach standby cache.", output)
+
+    detached_cache = next(cache for cache in TestRun.dut.cache_list if cache.cache_id == cache_id)
+    detached_cache.cache_device = None
+
     return output
 
 
@@ -510,6 +588,10 @@ def standby_activate_cache(cache_dev: Device, cache_id: int, shortcut: bool = Fa
     )
     if output.exit_code != 0:
         raise CmdException("Failed to activate standby cache.", output)
+
+    activated_cache = next(cache for cache in TestRun.dut.cache_list if cache.cache_id == cache_id)
+    activated_cache.cache_device = cache_dev
+
     return output