tests: minor refactor

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

test/functional/tests/cache_ops/test_cleaning_policy_operation.py

@@ -106,7 +106,7 @@ def test_cleaning_policies_in_write_back(cleaning_policy: CleaningPolicy):
 @pytest.mark.parametrize("cleaning_policy", CleaningPolicy)
 @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
 @pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
-def test_cleaning_policies_in_write_through(cleaning_policy):
+def test_cleaning_policies_in_write_through(cleaning_policy: CleaningPolicy):
     """
     title: Test for cleaning policy operation in Write-Through cache mode.
     description: |

test/functional/tests/cli/test_cli_help_spelling.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2022 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -12,23 +13,22 @@
 def test_cli_help_spelling():
     """
     title: Spelling test for 'help' command
-    description: Validates spelling of 'help' in CLI
+    description: |
+        Validates spelling of 'help' in CLI
     pass criteria:
-    - no spelling mistakes are found
+      - no spelling mistakes are found
     """
 
     cas_dictionary = os.path.join(TestRun.usr.repo_dir, "test", "functional", "resources")
 
     with TestRun.step("Run aspell"):
-        TestRun.executor.rsync_to(
-            f"{cas_dictionary}/",
-            f"{TestRun.usr.working_dir}/",
-            delete=True)
+        TestRun.executor.rsync_to(f"{cas_dictionary}/", f"{TestRun.usr.working_dir}/", delete=True)
         cas_dictionary = os.path.join(TestRun.usr.working_dir, "cas_ex.en.pws")
 
         output = TestRun.executor.run_expect_success(
             f"{casadm_bin} -H 2>&1 | aspell list -c --lang=en_US "
-            f"--add-extra-dicts={cas_dictionary}")
+            f"--add-extra-dicts={cas_dictionary}"
+        )
 
         if output.stdout:
             TestRun.LOGGER.error("Misspelled words found:\n")
@@ -37,13 +37,15 @@ def test_cli_help_spelling():
         output = TestRun.executor.run_expect_success(
             f"{casadm_bin} -H"
             " | awk '/Available commands:/{ cmd=1;next } /For detailed help/ { cmd=0 } "
-            "cmd { print $0 }' | grep -o '\\-\\-\\S*'")
+            "cmd { print $0 }' | grep -o '\\-\\-\\S*'"
+        )
         commands = output.stdout.splitlines()
 
         for command in commands:
             output = TestRun.executor.run_expect_success(
                 f"{casadm_bin} {command} -H | aspell list --lang=en_US "
-                f"--add-extra-dicts={cas_dictionary}")
+                f"--add-extra-dicts={cas_dictionary}"
+            )
 
             if output.stdout:
                 TestRun.LOGGER.error(f"Misspelled word found in command :{command}\n")

test/functional/tests/cli/test_cli_script.py

@@ -1,12 +1,12 @@
 #
 # Copyright(c) 2020-2021 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
-
 import pytest
 
-from api.cas import casadm, casadm_parser
+from api.cas import casadm
 from core.test_run import TestRun
 from test_utils.os_utils import sync
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
@@ -19,13 +19,14 @@
 @pytest.mark.parametrize("purge_target", ["cache", "core"])
 def test_purge(purge_target):
     """
-        title: Call purge without and with `--script` switch
-        description: |
-          Check if purge is called only when `--script` switch is used.
-        pass_criteria:
-          - casadm returns an error when `--script` is missing
-          - cache is wiped when purge command is used properly
+    title: Call purge without and with `--script` switch
+    description: |
+      Check if purge is called only when `--script` switch is used.
+    pass_criteria:
+      - casadm returns an error when `--script` is missing
+      - cache is wiped when purge command is used properly
     """
+
     with TestRun.step("Prepare devices"):
         cache_device = TestRun.disks["cache"]
         core_device = TestRun.disks["core"]
@@ -52,25 +53,19 @@ def test_purge(purge_target):
         dd.run()
         sync()
 
-    with TestRun.step(
-        f"Try to call purge-{purge_target} without `--script` switch"
-    ):
+    with TestRun.step(f"Try to call purge-{purge_target} without `--script` switch"):
         original_occupancy = cache.get_statistics().usage_stats.occupancy
         purge_params = f"--cache-id {cache.cache_id} "
         if purge_target == "core":
             purge_params += f"--core-id {core.core_id}"
-        TestRun.executor.run_expect_fail(
-            f"casadm --purge-{purge_target} {purge_params}"
-        )
+        TestRun.executor.run_expect_fail(f"casadm --purge-{purge_target} {purge_params}")
 
         if cache.get_statistics().usage_stats.occupancy != original_occupancy:
             TestRun.fail(
                 f"Purge {purge_target} should not be possible to use without `--script` switch!"
             )
 
-    with TestRun.step(
-        f"Try to call purge-{purge_target} with `--script` switch"
-    ):
+    with TestRun.step(f"Try to call purge-{purge_target} with `--script` switch"):
         TestRun.executor.run_expect_success(
             f"casadm --script --purge-{purge_target} {purge_params}"
         )
@@ -79,7 +74,5 @@ def test_purge(purge_target):
             TestRun.fail(f"{cache.get_statistics().usage_stats.occupancy.get_value()}")
             TestRun.fail(f"Purge {purge_target} should invalidate all cache lines!")
 
-    with TestRun.step(
-        f"Stop cache"
-    ):
+    with TestRun.step(f"Stop cache"):
         casadm.stop_all_caches()

test/functional/tests/cli/test_cli_start_stop.py

@@ -1,9 +1,11 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
 import pytest
+
 from random import randint
 
 from api.cas import casadm, casadm_parser, cli_messages
@@ -12,23 +14,22 @@
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_utils.size import Unit, Size
 
-CACHE_ID_RANGE = (1, 16384)
-CORE_ID_RANGE = (0, 4095)
-
 
 @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane]))
 @pytest.mark.parametrize("shortcut", [True, False])
 def test_cli_start_stop_default_id(shortcut):
     """
-        title: Test for starting a cache with a default ID - short and long command
-        description: |
-          Start a new cache with a default ID and then stop this cache.
-        pass_criteria:
-          - The cache has successfully started with default ID
-          - The cache has successfully stopped
+    title: Test for starting a cache with a default ID - short and long command
+    description: |
+        Start a new cache with a default ID and then stop this cache.
+    pass_criteria:
+      - The cache has successfully started with default ID
+      - The cache has successfully stopped
     """
+    CACHE_ID_RANGE = (1, 16384)
+
     with TestRun.step("Prepare the device for the cache."):
-        cache_device = TestRun.disks['cache']
+        cache_device = TestRun.disks["cache"]
         cache_device.create_partitions([Size(500, Unit.MebiByte)])
         cache_device = cache_device.partitions[0]
 
@@ -38,21 +39,27 @@ def test_cli_start_stop_default_id(shortcut):
     with TestRun.step("Check if the cache has started successfully."):
         caches = casadm_parser.get_caches()
         if len(caches) != 1:
-            TestRun.fail(f"There is a wrong number of caches found in the OS: {len(caches)}. "
-                         f"Should be only 1.")
+            TestRun.fail(
+                f"There is a wrong number of caches found in the OS: {len(caches)}. "
+                f"Should be only 1."
+            )
         if cache.cache_device.path != cache_device.path:
-            TestRun.fail(f"The cache has started using a wrong device:"
-                         f" {cache.cache_device.path}."
-                         f"\nShould use {cache_device.path}.")
+            TestRun.fail(
+                f"The cache has started using a wrong device:"
+                f" {cache.cache_device.path}."
+                f"\nShould use {cache_device.path}."
+            )
 
     with TestRun.step("Stop the cache."):
         casadm.stop_cache(cache.cache_id, shortcut=shortcut)
 
     with TestRun.step("Check if the cache has stopped properly."):
         caches = casadm_parser.get_caches()
         if len(caches) != 0:
-            TestRun.fail(f"There is a wrong number of caches found in the OS: {len(caches)}."
-                         f"\nNo cache should be present after stopping the cache.")
+            TestRun.fail(
+                f"There is a wrong number of caches found in the OS: {len(caches)}."
+                f"\nNo cache should be present after stopping the cache."
+            )
         output = casadm.list_caches(shortcut=shortcut)
         cli_messages.check_stdout_msg(output, cli_messages.no_caches_running)
 
@@ -61,15 +68,16 @@ def test_cli_start_stop_default_id(shortcut):
 @pytest.mark.parametrize("shortcut", [True, False])
 def test_cli_start_stop_custom_id(shortcut):
     """
-        title: Test for starting a cache with a custom ID - short and long command
-        description: |
-          Start a new cache with a random ID (from allowed pool) and then stop this cache.
-        pass_criteria:
-          - The cache has successfully started with a custom ID
-          - The cache has successfully stopped
+    title: Test for starting a cache with a custom ID - short and long command
+    description: |
+        Start a new cache with a random ID (from allowed pool) and then stop this cache.
+    pass_criteria:
+      - The cache has successfully started with a custom ID
+      - The cache has successfully stopped
     """
+
     with TestRun.step("Prepare the device for the cache."):
-        cache_device = TestRun.disks['cache']
+        cache_device = TestRun.disks["cache"]
         cache_device.create_partitions([Size(500, Unit.MebiByte)])
         cache_device = cache_device.partitions[0]
 
@@ -81,21 +89,27 @@ def test_cli_start_stop_custom_id(shortcut):
     with TestRun.step("Check if the cache has started successfully."):
         caches = casadm_parser.get_caches()
         if len(caches) != 1:
-            TestRun.fail(f"There is a wrong number of caches found in the OS: {len(caches)}. "
-                         f"Should be only 1.")
+            TestRun.fail(
+                f"There is a wrong number of caches found in the OS: {len(caches)}. "
+                f"Should be only 1."
+            )
         if cache.cache_device.path != cache_device.path:
-            TestRun.fail(f"The cache has started using a wrong device:"
-                         f" {cache.cache_device.path}."
-                         f"\nShould use {cache_device.path}.")
+            TestRun.fail(
+                f"The cache has started using a wrong device:"
+                f" {cache.cache_device.path}."
+                f"\nShould use {cache_device.path}."
+            )
 
     with TestRun.step("Stop the cache."):
         casadm.stop_cache(cache.cache_id, shortcut=shortcut)
 
     with TestRun.step("Check if the cache has stopped properly."):
         caches = casadm_parser.get_caches()
         if len(caches) != 0:
-            TestRun.fail(f"There is a wrong number of caches found in the OS: {len(caches)}."
-                         f"\nNo cache should be present after stopping the cache.")
+            TestRun.fail(
+                f"There is a wrong number of caches found in the OS: {len(caches)}."
+                f"\nNo cache should be present after stopping the cache."
+            )
         output = casadm.list_caches(shortcut=shortcut)
         cli_messages.check_stdout_msg(output, cli_messages.no_caches_running)
 
@@ -105,19 +119,20 @@ def test_cli_start_stop_custom_id(shortcut):
 @pytest.mark.parametrize("shortcut", [True, False])
 def test_cli_add_remove_default_id(shortcut):
     """
-        title: Test for adding and removing a core with a default ID - short and long command
-        description: |
-          Start a new cache and add a core to it without passing a core ID as an argument
-          and then remove this core from the cache.
-        pass_criteria:
-          - The core is added to the cache with a default ID
-          - The core is successfully removed from the cache
+    title: Test for adding and removing a core with a default ID - short and long command
+    description: |
+        Start a new cache and add a core to it without passing a core ID as an argument
+        and then remove this core from the cache.
+    pass_criteria:
+      - The core is added to the cache with a default ID
+      - The core is successfully removed from the cache
     """
+
     with TestRun.step("Prepare the devices."):
-        cache_disk = TestRun.disks['cache']
+        cache_disk = TestRun.disks["cache"]
         cache_disk.create_partitions([Size(50, Unit.MebiByte)])
         cache_device = cache_disk.partitions[0]
-        core_device = TestRun.disks['core']
+        core_device = TestRun.disks["core"]
 
     with TestRun.step("Start the cache and add the core."):
         cache = casadm.start_cache(cache_device, shortcut=shortcut, force=True)
@@ -156,19 +171,21 @@ def test_cli_add_remove_default_id(shortcut):
 @pytest.mark.parametrize("shortcut", [True, False])
 def test_cli_add_remove_custom_id(shortcut):
     """
-        title: Test for adding and removing a core with a custom ID - short and long command
-        description: |
-          Start a new cache and add a core to it with passing a random core ID
-          (from allowed pool) as an argument and then remove this core from the cache.
-        pass_criteria:
-          - The core is added to the cache with a default ID
-          - The core is successfully removed from the cache
+    title: Test for adding and removing a core with a custom ID - short and long command
+    description: |
+        Start a new cache and add a core to it with passing a random core ID
+        (from allowed pool) as an argument and then remove this core from the cache.
+    pass_criteria:
+      - The core is added to the cache with a default ID
+      - The core is successfully removed from the cache
     """
+    CORE_ID_RANGE = (0, 4095)
+
     with TestRun.step("Prepare the devices."):
-        cache_disk = TestRun.disks['cache']
+        cache_disk = TestRun.disks["cache"]
         cache_disk.create_partitions([Size(50, Unit.MebiByte)])
         cache_device = cache_disk.partitions[0]
-        core_device = TestRun.disks['core']
+        core_device = TestRun.disks["core"]
 
     with TestRun.step("Start the cache and add the core with a random ID."):
         core_id = randint(*CORE_ID_RANGE)
@@ -208,16 +225,17 @@ def test_cli_add_remove_custom_id(shortcut):
 @pytest.mark.parametrize("shortcut", [True, False])
 def test_cli_load_and_force(shortcut):
     """
-        title: Test if it is possible to use start command with 'load' and 'force' flag at once
-        description: |
-          Try to start cache with 'load' and 'force' options at the same time
-          and check if it is not possible to do
-        pass_criteria:
-          - Start cache command with both 'force' and 'load' options should fail
-          - Proper message should be received
+    title: Test if it is possible to use start command with 'load' and 'force' flag at once
+    description: |
+        Try to start cache with 'load' and 'force' options at the same time
+        and check if it is not possible to do
+    pass_criteria:
+      - Start cache command with both 'force' and 'load' options should fail
+      - Proper message should be received
     """
+
     with TestRun.step("Prepare cache."):
-        cache_device = TestRun.disks['cache']
+        cache_device = TestRun.disks["cache"]
         cache_device.create_partitions([Size(50, Unit.MebiByte)])
         cache_device = cache_device.partitions[0]
         cache = casadm.start_cache(cache_device)