tests: fix pep8 style violation

test/functional/api/cas/cache.py

@@ -50,7 +50,7 @@ def __get_cache_device(self) -> Device | None:
         if not cache:
             return None
 
-        if cache["device_path"] is "-":
+        if cache["device_path"] == "-":
             return None
 
         return Device(path=cache["device_path"])

test/functional/api/cas/cas_packaging.py

@@ -1,6 +1,6 @@
 #
 # Copyright(c) 2022 Intel Corporation
-# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# Copyright(c) 2024-2025 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -49,7 +49,7 @@ def create(
         arch: str = "",
         source: bool = False,
     ):
-        TestRun.LOGGER.info(f"Creating Open CAS RPM packages")
+        TestRun.LOGGER.info("Creating Open CAS RPM packages")
 
         self.packages_dir = (
             packages_dir or self.packages_dir or os.path.join(sources_dir, "packages")
@@ -79,7 +79,7 @@ def create(
         arch: str = "",
         source: bool = False,
     ):
-        TestRun.LOGGER.info(f"Creating Open CAS DEB packages")
+        TestRun.LOGGER.info("Creating Open CAS DEB packages")
 
         self.packages_dir = (
             packages_dir or self.packages_dir or os.path.join(sources_dir, "packages")

test/functional/api/cas/cli.py

@@ -447,7 +447,7 @@ def script_try_add_cmd(cache_id: str, core_dev: str, core_id: str) -> str:
     command = " --script --add-core --try-add"
     command += " --cache-id " + cache_id
     command += " --core-device " + core_dev
-    command += f" --core-id " + core_id
+    command += "--core-id " + core_id
     return casadm_bin + command
 
 

test/functional/tests/512b/test_different_io_sizes_support.py

@@ -55,7 +55,7 @@ def test_support_different_io_size(cache_mode):
         core.create_filesystem(Filesystem.xfs)
         core.mount(mountpoint)
 
-    with TestRun.step(f"Run fio"):
+    with TestRun.step("Run fio"):
         bs_list = [Size(x, Unit.KibiByte) for x in block_sizes]
 
         fio = (

test/functional/tests/basic/test_basic.py

@@ -51,7 +51,7 @@ def test_cas_version():
 
     with TestRun.step("Compare cmd and file versions"):
         if not all(file_cas_version == cmd_cas_version for cmd_cas_version in cmd_cas_versions):
-            TestRun.LOGGER.error(f"Cmd and file versions doesn't match")
+            TestRun.LOGGER.error("Cmd and file versions doesn't match")
 
 
 @pytest.mark.CI

test/functional/tests/cache_ops/test_attach_detach.py

@@ -14,7 +14,6 @@
 from api.cas.cli import attach_cache_cmd
 from api.cas.cli_messages import check_stderr_msg, attach_with_existing_metadata
 from connection.utils.output import CmdException
-from core.test_run import TestRun
 from core.test_run_utils import TestRun
 from storage_devices.disk import DiskTypeSet, DiskType, DiskTypeLowerThan
 from storage_devices.nullblk import NullBlk
@@ -225,7 +224,7 @@ def test_stop_cache_during_attach(cache_mode):
         core_dev.create_partitions([Size(2, Unit.GibiByte)])
         core_dev = core_dev.partitions[0]
 
-    with TestRun.step(f"Start cache and add core"):
+    with TestRun.step("Start cache and add core"):
         cache = casadm.start_cache(cache_dev, force=True, cache_mode=cache_mode)
         cache.add_core(core_dev)
 
@@ -259,4 +258,4 @@ def _get_random_uniq_cache_line_size(cache_line_size) -> CacheLineSize:
 
 
 def _get_random_uniq_cache_mode(cache_mode) -> CacheMode:
-    return random.choice([c for c in list(CacheMode) if c is not cache_mode])
+    return random.choice([c for c in list(CacheMode) if c is not cache_mode])

test/functional/tests/cache_ops/test_dynamic_cache_mode_switching.py

@@ -61,7 +61,7 @@ def test_cache_stop_and_load(cache_mode):
         core_dev = TestRun.disks["core"]
         core_dev.create_partitions([Size(2, Unit.GibiByte)])
 
-    with TestRun.step(f"Disable udev"):
+    with TestRun.step("Disable udev"):
         Udev.disable()
 
     with TestRun.step(f"Start cache in {cache_mode[0]} mode"):

test/functional/tests/cache_ops/test_multistream_seq_cutoff.py

@@ -57,7 +57,7 @@ def test_multistream_seq_cutoff_functional(streams_number, threshold):
     with TestRun.step("Disable udev"):
         Udev.disable()
 
-    with TestRun.step(f"Start cache in Write-Back cache mode"):
+    with TestRun.step("Start cache in Write-Back cache mode"):
         cache_disk = TestRun.disks["cache"]
         core_disk = TestRun.disks["core"]
         cache = casadm.start_cache(cache_disk, CacheMode.WB, force=True)
@@ -154,16 +154,16 @@ def test_multistream_seq_cutoff_stress_raw(streams_seq_rand):
 
         cache_disk.create_partitions([Size(1.5, Unit.GibiByte)])
 
-    with TestRun.step(f"Disable udev"):
+    with TestRun.step("Disable udev"):
         Udev.disable()
 
-    with TestRun.step(f"Start cache in Write-Back mode and add core"):
+    with TestRun.step("Start cache in Write-Back mode and add core"):
         cache = casadm.start_cache(
             cache_dev=cache_disk.partitions[0], cache_mode=CacheMode.WB, force=True
         )
         core = cache.add_core(core_disk)
 
-    with TestRun.step(f"Set seq-cutoff policy to always and threshold to 512KiB"):
+    with TestRun.step("Set seq-cutoff policy to always and threshold to 512KiB"):
         core.set_seq_cutoff_policy(SeqCutOffPolicy.always)
         core.set_seq_cutoff_threshold(Size(512, Unit.KibiByte))
 

test/functional/tests/cache_ops/test_seq_cutoff.py

@@ -312,7 +312,7 @@ def test_seq_cutoff_thresh(cache_line_size, io_dir, policy, verify_type):
     with TestRun.step("Disable udev"):
         Udev.disable()
 
-    with TestRun.step(f"Start cache and add core"):
+    with TestRun.step("Start cache and add core"):
         cache = casadm.start_cache(
             cache_dev=cache_part,
             force=True,
@@ -396,7 +396,7 @@ def test_seq_cutoff_thresh_fill(cache_line_size, io_dir):
     with TestRun.step("Disable udev"):
         Udev.disable()
 
-    with TestRun.step(f"Start cache and add core"):
+    with TestRun.step("Start cache and add core"):
         cache = casadm.start_cache(
             cache_dev=cache_part,
             force=True,

test/functional/tests/cli/test_cleaning_policy.py

@@ -9,7 +9,6 @@
 
 from core.test_run_utils import TestRun
 from type_def.size import Size, Unit
-from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools.fio.fio import Fio
 from test_tools.fio.fio_param import ReadWrite, IoEngine
@@ -47,7 +46,7 @@ def test_cleaning_policy_change():
     with TestRun.step("Disable udev"):
         Udev.disable()
 
-    with TestRun.step(f"Start cache in Write-Back mode and set cleaning policy to NOP"):
+    with TestRun.step("Start cache in Write-Back mode and set cleaning policy to NOP"):
         cache = casadm.start_cache(cache_dev, cache_mode=CacheMode.WB, force=True)
         cache.set_cleaning_policy(CleaningPolicy.nop)
 

test/functional/tests/cli/test_cli_detached_cache.py

@@ -233,7 +233,7 @@ def test_detached_cache_permitted_cli_management():
         tested_cmd = cli.attach_cache_cmd(cache_device.path, str(cache.cache_id))
         TestRun.LOGGER.info(f"Verify {tested_cmd}")
         TestRun.executor.run_expect_success(tested_cmd)
-        TestRun.LOGGER.info(f"Detach cache after successfully executed attach command")
+        TestRun.LOGGER.info("Detach cache after successfully executed attach command")
         cache.detach()
 
         tested_cmd = cli.stop_cmd(str(cache.cache_id))

test/functional/tests/cli/test_cli_standby.py

@@ -371,7 +371,7 @@ def test_activate_without_detach():
     with TestRun.step("Verify if cache is in standby state after failed activation"):
         caches = casadm_parser.get_caches()
         if len(caches) < 1:
-            TestRun.LOGGER.error(f'Cache not present in system')
+            TestRun.LOGGER.error("Cache not present in system")
         else:
             cache_status = caches[0].get_status()
             if cache_status != CacheStatus.standby:
@@ -453,7 +453,7 @@ def test_activate_neg_cache_line_size():
     with TestRun.step("Detach standby cache instance"):
         standby_cache.standby_detach()
 
-    with TestRun.step(f"Copy changed metadata to the standby instance"):
+    with TestRun.step("Copy changed metadata to the standby instance"):
         Dd().input(md_dump.full_path).output(standby_cache_dev.path).run()
         sync()
 
@@ -635,7 +635,7 @@ def test_standby_activate_with_corepool():
                                             cache_line_size=cache_line_size,
                                             force=True)
 
-    with TestRun.step(f"Copy changed metadata to the standby instance"):
+    with TestRun.step("Copy changed metadata to the standby instance"):
         Dd().input(active_cache_dev.path).output(f"/dev/{cache_exp_obj_name}").run()
 
     with TestRun.step("Detach standby cache instance"):

test/functional/tests/cli/test_manual_flush.py

@@ -10,7 +10,6 @@
 from core.test_run_utils import TestRun
 from storage_devices.device import Device
 from type_def.size import Size, Unit
-from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools.fio.fio import Fio
 from test_tools.fio.fio_param import ReadWrite, IoEngine

test/functional/tests/cli/test_seq_cutoff_settings.py

@@ -130,9 +130,9 @@ def test_seq_cutoff_set_get_policy_cache(policy):
         cache.set_seq_cutoff_policy(policy)
 
     for i in TestRun.iteration(range(0, len(cores)), "Verifying if proper policy was set"):
-        with TestRun.step(f"Check if proper sequential cutoff policy was set for core"):
+        with TestRun.step("Check if proper sequential cutoff policy was set for core"):
             if cores[i].get_seq_cut_off_policy() != policy:
-                TestRun.fail(f"Wrong core sequential cutoff policy: "
+                TestRun.fail("Wrong core sequential cutoff policy: "
                              f"{cores[i].get_seq_cut_off_policy()} "
                              f"should be {policy}")
 
@@ -186,7 +186,7 @@ def test_seq_cutoff_policy_load():
             enumerate(cores[:-1]),
             "Check if proper policies have been loaded"
     ):
-        with TestRun.step(f"Check if proper sequential cutoff policy was loaded"):
+        with TestRun.step("Check if proper sequential cutoff policy was loaded"):
             if cores[i].get_seq_cut_off_policy() != policies[i]:
                 TestRun.fail(f"Wrong sequential cutoff policy loaded: "
                              f"{cores[i].get_seq_cut_off_policy()} "
@@ -239,8 +239,8 @@ def test_seq_cutoff_set_invalid_threshold(threshold):
                 not in output.stderr:
             TestRun.fail("Command succeeded (should fail)!")
 
-    with TestRun.step(f"Setting cache sequential cutoff threshold "
-                      f"to value passed as a float"):
+    with TestRun.step("Setting cache sequential cutoff threshold "
+                      "to value passed as a float"):
         command = set_param_cutoff_cmd(
             cache_id=str(cache.cache_id), core_id=str(core.core_id),
             threshold=str(_threshold.get_value(Unit.KiloByte)))

test/functional/tests/conftest.py

@@ -212,9 +212,9 @@ def base_prepare(item):
             create_partition_table(disk, PartitionTable.gpt)
 
         TestRun.usr.already_updated = True
-        TestRun.LOGGER.add_build_info(f"Commit hash:")
+        TestRun.LOGGER.add_build_info("Commit hash:")
         TestRun.LOGGER.add_build_info(f"{git.get_current_commit_hash()}")
-        TestRun.LOGGER.add_build_info(f"Commit message:")
+        TestRun.LOGGER.add_build_info("Commit message:")
         TestRun.LOGGER.add_build_info(f"{git.get_current_commit_message()}")
 
 

test/functional/tests/data_integrity/test_data_integrity_12h.py

@@ -1,6 +1,6 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
-# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# Copyright(c) 2024-2025 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -57,9 +57,9 @@ def prepare(cache_mode):
     cache_device = cache_device.partitions[0]
     core_device = core_device.partitions[0]
 
-    TestRun.LOGGER.info(f"Starting cache")
+    TestRun.LOGGER.info("Starting cache")
     cache = casadm.start_cache(cache_device, cache_mode, force=True)
-    TestRun.LOGGER.info(f"Adding core device")
+    TestRun.LOGGER.info("Adding core device")
     core = casadm.add_core(cache, core_dev=core_device)
 
     return cache, core

test/functional/tests/data_integrity/test_data_integrity_unplug.py

@@ -271,7 +271,7 @@ def gen_csums(dev_path, seqno, pattern, csums, csums_rev):
 
     for j in range(num_jobs):
         for b in range(job_workset_blocks):
-            if pattern[j][b] != -1 and not pattern[j][b] in csums[j]:
+            if pattern[j][b] != -1 and pattern[j][b] not in csums[j]:
                 csums[j][pattern[j][b]] = cs[j][b]
                 csums_rev[cs[j][b]] = get_data_name(j, b, pattern[j][b])
 

test/functional/tests/example/example_test.py

@@ -1,6 +1,6 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
-# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# Copyright(c) 2024-2025 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -293,4 +293,4 @@ def test_lvm_example():
             if lvm not in lvms:
                 TestRun.LOGGER.error(f"Created LVM {lvm.volume_name} not discovered in system!")
 
-        TestRun.LOGGER.info(f"Created LVMs present in the system.")
+        TestRun.LOGGER.info("Created LVMs present in the system.")

test/functional/tests/failover_standby/test_cache_activation_perf.py

@@ -1,6 +1,6 @@
 #
 # Copyright(c) 2022 Intel Corporation
-# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# Copyright(c) 2024-2025 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 from datetime import timedelta, datetime
@@ -66,7 +66,7 @@ def test_cache_activation_time(prefill):
         cache.set_seq_cutoff_policy(SeqCutOffPolicy.never)
         cache.set_cleaning_policy(CleaningPolicy.nop)
 
-    with TestRun.step(f"Pre-filling the cache with dirty data to target threshold"):
+    with TestRun.step("Pre-filling the cache with dirty data to target threshold"):
         if prefill is True:
             TestRun.LOGGER.info(f"Target prefill threshold: {prefill_threshold}%")
 
@@ -85,7 +85,7 @@ def test_cache_activation_time(prefill):
                 .run()
             )
         else:
-            TestRun.LOGGER.info(f"Target prefill threshold: 0% - prefill skipped")
+            TestRun.LOGGER.info("Target prefill threshold: 0% - prefill skipped")
 
     with TestRun.step("Check usage statistics after pre-fill"):
         stats = cache.get_statistics(percentage_val=True)
@@ -116,7 +116,7 @@ def test_cache_activation_time(prefill):
                                             cache_id=current_cache_id)
 
     with TestRun.step("verify cache exported object has appeared"):
-        output = TestRun.executor.run_expect_success(f"ls -la /dev/ | grep cas-cache-1")
+        output = TestRun.executor.run_expect_success("ls -la /dev/ | grep cas-cache-1")
         if output.stdout[0] != "b":
             TestRun.fail("The cache exported object is not a block device")
 
@@ -135,7 +135,7 @@ def test_cache_activation_time(prefill):
         standby_cache.standby_detach()
 
     with TestRun.step("Verify exp. obj. disappeared"):
-        TestRun.executor.run_expect_fail(f"ls -la /dev/ | grep cas-cache-1")
+        TestRun.executor.run_expect_fail("ls -la /dev/ | grep cas-cache-1")
 
     with TestRun.step("Activate passive cache and measure the activation time"):
         start_time = datetime.now()