Minor refactor

test/functional/api/cas/cache.py

@@ -32,7 +32,7 @@ def __get_cache_id(self) -> int:
     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) -> list:
+    def get_cores(self) -> list:
         return get_cores(self.cache_id)
 
     def get_cache_line_size(self) -> CacheLineSize:

test/functional/tests/cli/test_cli_start_stop.py

@@ -1,11 +1,13 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
-import pytest
 from random import randint
 
+import pytest
+
 from api.cas import casadm, casadm_parser, cli_messages
 from api.cas.cli import start_cmd
 from core.test_run import TestRun
@@ -125,9 +127,9 @@ def test_cli_add_remove_default_id(shortcut):
 
     with TestRun.step("Check if the core is added to the cache."):
         caches = casadm_parser.get_caches()
-        if len(caches[0].get_core_devices()) != 1:
+        if len(caches[0].get_cores()) != 1:
             TestRun.fail("One core should be present in the cache.")
-        if caches[0].get_core_devices()[0].path != core.path:
+        if caches[0].get_cores()[0].path != core.path:
             TestRun.fail("The core path should be equal to the path of the core added.")
 
     with TestRun.step("Remove the core from the cache."):
@@ -137,7 +139,7 @@ def test_cli_add_remove_default_id(shortcut):
         caches = casadm_parser.get_caches()
         if len(caches) != 1:
             TestRun.fail("One cache should be still present after removing the core.")
-        if len(caches[0].get_core_devices()) != 0:
+        if len(caches[0].get_cores()) != 0:
             TestRun.fail("No core device should be present after removing the core.")
 
     with TestRun.step("Stop the cache."):
@@ -178,9 +180,9 @@ def test_cli_add_remove_custom_id(shortcut):
 
     with TestRun.step("Check if the core is added to the cache."):
         caches = casadm_parser.get_caches()
-        if len(caches[0].get_core_devices()) != 1:
+        if len(caches[0].get_cores()) != 1:
             TestRun.fail("One core should be present in the cache.")
-        if caches[0].get_core_devices()[0].path != core.path:
+        if caches[0].get_cores()[0].path != core.path:
             TestRun.fail("The core path should be equal to the path of the core added.")
 
     with TestRun.step("Remove the core from the cache."):
@@ -190,7 +192,7 @@ def test_cli_add_remove_custom_id(shortcut):
         caches = casadm_parser.get_caches()
         if len(caches) != 1:
             TestRun.fail("One cache should be still present after removing the core.")
-        if len(caches[0].get_core_devices()) != 0:
+        if len(caches[0].get_cores()) != 0:
             TestRun.fail("No core device should be present after removing the core.")
 
     with TestRun.step("Stop the cache."):

test/functional/tests/cli/test_seq_cutoff_settings.py

@@ -1,18 +1,20 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
 
-import pytest
 import random
 from ctypes import c_uint32
+
+import pytest
+
 from api.cas import casadm
 from api.cas.cache_config import SeqCutOffPolicy
-from api.cas.core import SEQ_CUTOFF_THRESHOLD_MAX, SEQ_CUT_OFF_THRESHOLD_DEFAULT
 from api.cas.casadm import set_param_cutoff_cmd
+from api.cas.core import SEQ_CUTOFF_THRESHOLD_MAX, SEQ_CUT_OFF_THRESHOLD_DEFAULT
 from core.test_run import TestRun
-
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_utils.size import Size, Unit
 
@@ -137,7 +139,7 @@ def test_seq_cutoff_policy_load():
         loaded_cache = casadm.load_cache(cache.cache_device)
 
     with TestRun.step("Getting cores from loaded cache"):
-        cores = loaded_cache.get_core_devices()
+        cores = loaded_cache.get_cores()
 
     for i, core in TestRun.iteration(enumerate(cores[:-1]), "Check if proper policies have "
                                                             "been loaded"):
@@ -251,7 +253,7 @@ def test_seq_cutoff_threshold_load(threshold):
         loaded_cache = casadm.load_cache(cache.cache_device)
 
     with TestRun.step("Getting core from loaded cache"):
-        cores_load = loaded_cache.get_core_devices()
+        cores_load = loaded_cache.get_cores()
 
     with TestRun.step("Check if proper sequential cut off policy was loaded"):
         if cores_load[0].get_seq_cut_off_threshold() != _threshold:

test/functional/tests/conftest.py

@@ -14,7 +14,7 @@
 
 from datetime import timedelta
 
-sys.path.append(os.path.join(os.path.dirname(__file__), "../test-framework"))
+sys.path.append(os.path.join(os.path.dirname(__file__), "..", "test-framework"))
 
 from core.test_run import Blocked
 from core.test_run_utils import TestRun
@@ -97,7 +97,7 @@ def pytest_runtest_setup(item):
                 f"{ex}\nYou need to specify DUT config. See the example_dut_config.py file"
             )
 
-        dut_config["plugins_dir"] = os.path.join(os.path.dirname(__file__), "../lib")
+        dut_config["plugins_dir"] = os.path.join(os.path.dirname(__file__), "..", "lib")
         dut_config["opt_plugins"] = {"test_wrapper": {}, "serial_log": {}, "power_control": {}}
         dut_config["extra_logs"] = {"cas": "/var/log/opencas.log"}
 
@@ -122,7 +122,7 @@ def pytest_runtest_setup(item):
             )
 
         TestRun.usr = Opencas(
-            repo_dir=os.path.join(os.path.dirname(__file__), "../../.."),
+            repo_dir=os.path.join(os.path.dirname(__file__), "..", "..", ".."),
             working_dir=dut_config["working_dir"],
         )
         if item.config.getoption("--fuzzy-iter-count"):
@@ -291,7 +291,7 @@ def __drbd_cleanup():
     from storage_devices.drbd import Drbd
 
     Drbd.down_all()
-    # If drbd instance had been configured on top of the CAS, the previos attempt to stop
+    # If drbd instance had been configured on top of the CAS, the previous attempt to stop
     # failed. As drbd has been stopped try to stop CAS one more time.
     if installer.check_if_installed():
         casadm.stop_all_caches()

test/functional/tests/incremental_load/test_inactive_cores.py

@@ -143,7 +143,7 @@ def test_core_inactive_stats_usage():
         active_cores_clean_stats = 0
         active_cores_dirty_stats = 0
 
-        active_cores = cache.get_core_devices()
+        active_cores = cache.get_cores()
         for core in active_cores:
             core_stats = core.get_statistics()
             active_cores_occupancy_stats += core_stats.usage_stats.occupancy

test/functional/tests/incremental_load/test_incremental_load.py

@@ -4,11 +4,11 @@
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
-import pytest
 import time
-
 from random import shuffle
 
+import pytest
+
 from api.cas import casadm, cli, cli_messages
 from api.cas.cache_config import (
     CacheStatus,
@@ -128,7 +128,7 @@ def test_incremental_load_missing_core_device():
             if core.get_status() is not CoreStatus.active:
                 TestRun.fail(f"Core {core.core_id} should be active but is {core.get_status()}.")
 
-        core_with_missing_device = cache.get_core_devices()[-1]
+        core_with_missing_device = cache.get_cores()[-1]
 
     with TestRun.step("Stop cache."):
         cache.stop()
@@ -142,7 +142,7 @@ def test_incremental_load_missing_core_device():
             TestRun.fail(
                 f"Cache {cache.cache_id} should be incomplete but is " f"{cache.get_status()}."
             )
-        for core in cache.get_core_devices():
+        for core in cache.get_cores():
             if core.get_status() is not CoreStatus.active:
                 TestRun.fail(f"Core {core.core_id} should be Active but is {core.get_status()}.")
         if core_with_missing_device.get_status() is not CoreStatus.inactive:

test/functional/tests/initialize/test_negative_load.py

@@ -1,10 +1,12 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
 
 import pytest
+
 from api.cas import casadm, casadm_parser, cli, cli_messages
 from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
@@ -56,7 +58,7 @@ def test_load_occupied_id():
         if caches[0].cache_id != 1:
             TestRun.LOGGER.error("Wrong cache id.")
 
-        cores = caches[0].get_core_devices()
+        cores = caches[0].get_cores()
         if len(cores) != 0:
             TestRun.LOGGER.error("Inappropriate number of cores after load!")
 

test/functional/tests/initialize/test_simulation_startup.py

@@ -65,7 +65,7 @@ def test_simulation_startup_from_config():
             )
 
     with TestRun.step("Verify if core is working"):
-        core = cache.get_core_devices()[0]
+        core = cache.get_cores()[0]
         if core.get_status() is not CoreStatus.active:
             TestRun.fail(
                 f"Core {core.core_id} should be active but is in {core.get_status()} " f"state."
@@ -89,7 +89,7 @@ def test_simulation_startup_from_config():
             )
 
     with TestRun.step("Verify if core is working"):
-        cores = cache.get_core_devices()
+        cores = cache.get_cores()
         if not cores:
             TestRun.fail("Core is not working")
         core = cores[0]

test/functional/tests/stats/test_ioclass_stats.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies
 # SPDX-License-Identifier: BSD-3-Clause
 #
 import random
@@ -17,7 +18,6 @@
 )
 from api.cas.statistics import (
     config_stats_ioclass,
-    usage_stats,
     usage_stats_ioclass,
     request_stats,
     block_stats_core,
@@ -234,7 +234,7 @@ def test_ioclass_stats_sections(stat_filter, per_core, random_cls):
                       f"{'cores' if per_core else 'caches'}"):
         for cache in caches:
             with TestRun.group(f"Cache {cache.cache_id}"):
-                for core in cache.get_core_devices():
+                for core in cache.get_cores():
                     if per_core:
                         TestRun.LOGGER.info(f"Core {core.cache_id}-{core.core_id}")
                     statistics = (
@@ -256,7 +256,7 @@ def test_ioclass_stats_sections(stat_filter, per_core, random_cls):
                       f"class for all {'cores' if per_core else 'caches'}"):
         for cache in caches:
             with TestRun.group(f"Cache {cache.cache_id}"):
-                for core in cache.get_core_devices():
+                for core in cache.get_cores():
                     core_info = f"Core {core.cache_id}-{core.core_id} ," if per_core else ""
                     for class_id in range(ioclass_config.MAX_IO_CLASS_ID + 1):
                         with TestRun.group(core_info + f"IO class id {class_id}"):