From 0c81b8a542ee5bc9b9f4d97bea26b7efd691993d Mon Sep 17 00:00:00 2001 From: Kamil Gierszewski Date: Tue, 10 Dec 2024 06:12:18 +0100 Subject: [PATCH] tests: minor refactor Signed-off-by: Kamil Gierszewski --- .../test_cleaning_policy_operation.py | 2 +- .../tests/cli/test_cli_help_spelling.py | 20 +-- test/functional/tests/cli/test_cli_script.py | 33 ++--- .../tests/cli/test_cli_start_stop.py | 132 ++++++++++-------- 4 files changed, 100 insertions(+), 87 deletions(-) diff --git a/test/functional/tests/cache_ops/test_cleaning_policy_operation.py b/test/functional/tests/cache_ops/test_cleaning_policy_operation.py index be6cf25b4..d9ac91cac 100644 --- a/test/functional/tests/cache_ops/test_cleaning_policy_operation.py +++ b/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: | diff --git a/test/functional/tests/cli/test_cli_help_spelling.py b/test/functional/tests/cli/test_cli_help_spelling.py index aacdbf84e..c3be5917f 100644 --- a/test/functional/tests/cli/test_cli_help_spelling.py +++ b/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") diff --git a/test/functional/tests/cli/test_cli_script.py b/test/functional/tests/cli/test_cli_script.py index 81217013e..2286b4fc6 100644 --- a/test/functional/tests/cli/test_cli_script.py +++ b/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() diff --git a/test/functional/tests/cli/test_cli_start_stop.py b/test/functional/tests/cli/test_cli_start_stop.py index 7f1b503d9..2edc1d979 100644 --- a/test/functional/tests/cli/test_cli_start_stop.py +++ b/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,12 +39,16 @@ 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) @@ -51,8 +56,10 @@ def test_cli_start_stop_default_id(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,12 +89,16 @@ 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) @@ -94,8 +106,10 @@ def test_cli_start_stop_custom_id(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)