# # 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, dmesg from api.cas.casadm import standby_init from api.cas.cli import casadm_bin, standby_init_cmd from core.test_run import TestRun from storage_devices.device import Device from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan from test_tools.dd import Dd from test_tools.disk_utils import Filesystem from test_utils.filesystem.file import File from test_utils.os_utils import sync from test_utils.output import CmdException from test_utils.size import Size, Unit from api.cas.cli_messages import ( check_stderr_msg, missing_param, disallowed_param, operation_forbiden_in_standby, mutually_exclusive_params_init, mutually_exclusive_params_load, activate_without_detach, cache_line_size_mismatch, start_cache_with_existing_metadata, standby_init_with_existing_filesystem, unexpected_cls_option, ) from api.cas.cache_config import CacheLineSize, CacheStatus from api.cas import cli from api.cas.ioclass_config import IoClass from api.cas.core import CoreStatus @pytest.mark.CI @pytest.mark.skip(reason="Standby mode is not supported") @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane])) def test_standby_neg_cli_params(): """ title: Verifying parameters for starting a standby cache instance description: | Try executing the standby init command with required arguments missing or disallowed arguments present. pass_criteria: - The execution is unsuccessful for all improper argument combinations - A proper error message is displayed for unsuccessful executions """ with TestRun.step("Prepare the device for the cache."): cache_device = TestRun.disks["cache"] cache_device.create_partitions([Size(500, Unit.MebiByte)]) cache_device = cache_device.partitions[0] with TestRun.step("Prepare config for testing standby init without required params"): init_required_params = dict( [("--cache-device", cache_device.path), ("--cache-id", 5), ("--cache-line-size", 32)] ) # Prepare full valid `standby init` command valid_cmd = casadm_bin + " --standby --init" for name, value in init_required_params.items(): valid_cmd += f" {name} {value}" # Try to initialize standby instance with one missing param at the time for name, value in init_required_params.items(): with TestRun.step(f'Try to init standby instance without "{name}" param'): tested_param = f"{name} {value}" tested_cmd = valid_cmd.replace(tested_param, "") output = TestRun.executor.run(tested_cmd) if output.exit_code == 0: TestRun.LOGGER.error( f'"{tested_cmd}" command succeeded despite missing required "{name}" parameter!' ) if not check_stderr_msg(output, missing_param) or name not in output.stderr: TestRun.LOGGER.error( f'Expected error message in format "{missing_param[0]}" with "{name}" ' f'(the missing param). Got "{output.stderr}" instead.' ) with TestRun.step("Prepare config for testing standby init with disallowed params"): init_disallowed_params = dict( [ ("--core-device", "/dev/disk/by-id/core_dev_id"), ("--core-id", 5), ("--cache-mode", 32), ("--file", "/etc/opencas/ioclass-config.csv"), ("--io-class-id", "0"), ] ) for name, value in init_disallowed_params.items(): with TestRun.step(f'Try to init standby instance with disallowed "{name}" param'): tested_param = f"{name} {value}" tested_cmd = f"{valid_cmd} {tested_param}" output = TestRun.executor.run(tested_cmd) if output.exit_code == 0: TestRun.LOGGER.error( f'"{tested_cmd}" command succeeded despite disallowed "{name}" parameter!' ) if not check_stderr_msg(output, disallowed_param): TestRun.LOGGER.error( f'Expected error message in format "{disallowed_param[0]}" ' f'Got "{output.stderr}" instead.' ) @pytest.mark.CI @pytest.mark.skip(reason="Standby mode is not supported") @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane])) def test_activate_neg_cli_params(): """ title: Verifying parameters for activating a standby cache instance. description: | Try executing the standby activate command with required arguments missing or disallowed arguments present. pass_criteria: -The execution is unsuccessful for all improper argument combinations -A proper error message is displayed for unsuccessful executions """ with TestRun.step("Prepare the device for the cache."): cache_device = TestRun.disks["cache"] cache_device.create_partitions([Size(500, Unit.MebiByte)]) cache_device = cache_device.partitions[0] cache_id = 1 with TestRun.step("Init standby cache"): cache_dev = Device(cache_device.path) cache = standby_init( cache_dev=cache_dev, cache_id=cache_id, cache_line_size=CacheLineSize.LINE_32KiB, force=True ) with TestRun.step("Detach standby cache"): cache.standby_detach() # Test standby activate with TestRun.step("Prepare config for testing standby activate with required params"): standby_activate_required_params = dict( [("--cache-device", cache_device.path), ("--cache-id", cache_id)] ) # Prepare full valid `standby activate` command valid_cmd = casadm_bin + " --standby --activate" for name, value in standby_activate_required_params.items(): valid_cmd += f" {name} {value}" for name, value in standby_activate_required_params.items(): with TestRun.step(f'Try to standby activate instance without "{name}" param'): tested_param = f"{name} {value}" tested_cmd = valid_cmd.replace(tested_param, "") output = TestRun.executor.run(tested_cmd) if output.exit_code == 0: TestRun.LOGGER.error( f'"{tested_cmd}" command succeeded despite missing obligatory' f' "{name}" parameter!' ) if not check_stderr_msg(output, missing_param) or name not in output.stderr: TestRun.LOGGER.error( f'Expected error message in format "{missing_param[0]}" with "{name}" ' f'(the missing param). Got "{output.stderr}" instead.' ) with TestRun.step("Prepare config for testing standby activate with disallowed params"): activate_disallowed_params = [ ("--cache-line-size", 32, unexpected_cls_option), ("--core-device", "/dev/disk/by-id/core_dev_id", disallowed_param), ("--core-id", 5, disallowed_param), ("--cache-mode", 32, disallowed_param), ("--file", "/etc/opencas/ioclass-config.csv", disallowed_param), ("--io-class-id", "0", disallowed_param), ] for name, value, expected_error_message in activate_disallowed_params: with TestRun.step(f'Try to activate standby instance with disallowed "{name}" param'): tested_param = f"{name} {value}" tested_cmd = f"{valid_cmd} {tested_param}" output = TestRun.executor.run(tested_cmd) if output.exit_code == 0: TestRun.LOGGER.error( f'"{tested_cmd}" command succeeded despite disallowed "{name}" parameter!' ) if not check_stderr_msg(output, expected_error_message): TestRun.LOGGER.error( f'Expected error message in format "{expected_error_message[0]}" ' f'Got "{output.stderr}" instead.' ) @pytest.mark.CI @pytest.mark.skip(reason="Standby mode is not supported") @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane])) def test_standby_neg_cli_management(): """ title: Blocking management commands in standby state description: | Try executing management commands for a cache in standby state pass_criteria: - The execution is unsuccessful for blocked management commands - The execution is successful for allowed management commands - A proper error message is displayed for unsuccessful executions """ with TestRun.step("Prepare the device for the cache."): device = TestRun.disks["cache"] device.create_partitions([Size(500, Unit.MebiByte), Size(500, Unit.MebiByte)]) cache_device = device.partitions[0] core_device = device.partitions[1] with TestRun.step("Prepare the standby instance"): cache_id = 1 cache = casadm.standby_init( cache_dev=cache_device, cache_id=cache_id, cache_line_size=CacheLineSize.LINE_32KiB, force=True ) ioclass_config_path = "/tmp/standby_cli_neg_mngt_test_ioclass_config_file.csv" TestRun.executor.run(f"rm -rf {ioclass_config_path}") random_ioclass_config = IoClass.generate_random_ioclass_list(5) IoClass.save_list_to_config_file( random_ioclass_config, ioclass_config_path=ioclass_config_path ) blocked_mngt_commands = [ cli.get_param_cutoff_cmd(str(cache_id), "1"), cli.get_param_cleaning_cmd(str(cache_id)), cli.get_param_cleaning_alru_cmd(str(cache_id)), cli.get_param_cleaning_acp_cmd(str(cache_id)), cli.set_param_cutoff_cmd(str(cache_id), "1", threshold="1"), cli.set_param_cutoff_cmd(str(cache_id), policy="never"), cli.set_param_cleaning_cmd(str(cache_id), policy="nop"), cli.set_param_cleaning_alru_cmd(str(cache_id), wake_up="30"), cli.set_param_cleaning_acp_cmd(str(cache_id), wake_up="100"), cli.set_param_promotion_cmd(str(cache_id), policy="nhit"), cli.set_param_promotion_nhit_cmd(str(cache_id), threshold="5"), cli.set_cache_mode_cmd("wb", str(cache_id)), cli.add_core_cmd(str(cache_id), core_device.path), cli.remove_core_cmd(str(cache_id), "1"), cli.remove_inactive_cmd(str(cache_id), "1"), cli.reset_counters_cmd(str(cache_id)), cli.flush_cache_cmd(str(cache_id)), cli.flush_core_cmd(str(cache_id), "1"), cli.load_io_classes_cmd(str(cache_id), ioclass_config_path), cli.list_io_classes_cmd(str(cache_id), output_format="csv"), cli.script_try_add_cmd(str(cache_id), core_device.path, core_id="1"), cli.script_purge_cache_cmd(str(cache_id)), cli.script_purge_core_cmd(str(cache_id), "1"), cli.script_detach_core_cmd(str(cache_id), "1"), cli.script_remove_core_cmd(str(cache_id), "1"), ] with TestRun.step("Try to execute forbidden management commands in standby mode"): for cmd in blocked_mngt_commands: TestRun.LOGGER.info(f"Verify {cmd}") output = TestRun.executor.run_expect_fail(cmd) if not check_stderr_msg(output, operation_forbiden_in_standby): TestRun.LOGGER.error( f'Expected the following error message "{operation_forbiden_in_standby[0]}" ' f'Got "{output.stderr}" instead.' ) with TestRun.step("Stop the standby instance"): TestRun.executor.run(f"rm -rf {ioclass_config_path}") cache.stop() @pytest.mark.CI @pytest.mark.skip(reason="Standby mode is not supported") @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane])) def test_start_neg_cli_flags(): """ title: Blocking standby start command with mutually exclusive flags description: | Try executing the standby start command with different combinations of mutually exclusive flags. pass_criteria: - The command execution is unsuccessful for commands with mutually exclusive flags - A proper error message is displayed """ with TestRun.step("Prepare the device for the cache."): cache_device = TestRun.disks["cache"] cache_device.create_partitions([Size(500, Unit.MebiByte)]) cache_device = cache_device.partitions[0] cache_id = 1 cache_line_size = 32 with TestRun.step("Try to start standby cache with mutually exclusive parameters"): init_required_params = f' --cache-device {cache_device.path}' \ f' --cache-id {cache_id}' \ f' --cache-line-size {cache_line_size}' mutually_exclusive_cmd_init = f"{casadm_bin} --standby --init --load" \ f" {init_required_params}" output = TestRun.executor.run_expect_fail(mutually_exclusive_cmd_init) if not check_stderr_msg(output, mutually_exclusive_params_init): TestRun.LOGGER.error( f'Expected error message in format ' f'"{mutually_exclusive_params_init[0]}"' f'Got "{output.stderr}" instead.' ) mutually_exclusive_cmd_load = [ f"{casadm_bin} --standby --load --cache-device {cache_device.path}" f" --cache-id {cache_id}", f"{casadm_bin} --standby --load --cache-device {cache_device.path}" f" --cache-line-size {cache_line_size}", f"{casadm_bin} --standby --load --cache-device {cache_device.path}" f" --force" ] for cmd in mutually_exclusive_cmd_load: output = TestRun.executor.run_expect_fail(cmd) if not check_stderr_msg(output, mutually_exclusive_params_load): TestRun.LOGGER.error( f'Expected error message in format ' f'"{mutually_exclusive_params_load[0]}"' f'Got "{output.stderr}" instead.' ) @pytest.mark.CI @pytest.mark.skip(reason="Standby mode is not supported") @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane])) def test_activate_without_detach(): """ title: Activate cache without detach command. description: | Try activate passive cache without detach command before activation. pass_criteria: - The activation is not possible - The cache remains in Standby state after unsuccessful activation - The cache exported object is present after an unsuccessful activation """ with TestRun.step("Prepare the device for the cache."): cache_dev = TestRun.disks["cache"] cache_dev.create_partitions([Size(500, Unit.MebiByte)]) cache_dev = cache_dev.partitions[0] cache_id = 1 cache_exp_obj_name = f"cas-cache-{cache_id}" with TestRun.step("Start cache instance."): cache = casadm.start_cache(cache_dev=cache_dev, cache_id=cache_id) with TestRun.step("Stop cache instance."): cache.stop() with TestRun.step("Load standby cache instance."): casadm.standby_load(cache_dev=cache_dev) with TestRun.step("Verify if the cache exported object appeared in the system"): output = TestRun.executor.run_expect_success(f"ls -la /dev/ | grep {cache_exp_obj_name}") if output.stdout[0] != "b": TestRun.fail("The cache exported object is not a block device") with TestRun.step("Try to activate cache instance"): cmd = f"{casadm_bin} --standby --activate --cache-id {cache_id} --cache-device " \ f"{cache_dev.path}" output = TestRun.executor.run(cmd) if not check_stderr_msg(output, activate_without_detach): TestRun.LOGGER.error( f'Expected error message in format ' f'"{activate_without_detach[0]}"' f'Got "{output.stderr}" instead.' ) 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') else: cache_status = caches[0].get_status() if cache_status != CacheStatus.standby: TestRun.LOGGER.error( f'Expected Cache state: "{CacheStatus.standby.value}" ' f'Got "{cache_status.value}" instead.' ) with TestRun.step("Verify if the cache exported object remains in the system"): output = TestRun.executor.run_expect_success(f"ls -la /dev/ | grep {cache_exp_obj_name}") if output.stdout[0] != "b": TestRun.fail("The cache exported object is not a block device") @pytest.mark.CI @pytest.mark.skip(reason="Standby mode is not supported") @pytest.mark.require_disk("active_cache", DiskTypeSet([DiskType.nand, DiskType.optane])) @pytest.mark.require_disk("standby_cache", DiskTypeSet([DiskType.nand, DiskType.optane])) def test_activate_neg_cache_line_size(): """ title: Blocking cache with mismatching cache line size activation. description: | Try restoring cache operations from a replicated cache that was initialized with different cache line size than the original cache. pass_criteria: - The activation is cancelled - The cache remains in Standby detached state after an unsuccessful activation - A proper error message is displayed """ with TestRun.step("Prepare cache devices"): active_cache_dev = TestRun.disks["active_cache"] active_cache_dev.create_partitions([Size(500, Unit.MebiByte)]) active_cache_dev = active_cache_dev.partitions[0] standby_cache_dev = TestRun.disks["standby_cache"] standby_cache_dev.create_partitions([Size(500, Unit.MebiByte)]) standby_cache_dev = standby_cache_dev.partitions[0] cache_id = 1 active_cls, standby_cls = CacheLineSize.LINE_4KiB, CacheLineSize.LINE_16KiB cache_exp_obj_name = f"cas-cache-{cache_id}" with TestRun.step("Start active cache instance."): active_cache = casadm.start_cache(cache_dev=active_cache_dev, cache_id=cache_id, cache_line_size=active_cls) with TestRun.step("Create dump file with cache metadata"): with TestRun.step("Get metadata size"): dmesg_out = TestRun.executor.run_expect_success("dmesg").stdout md_size = dmesg.get_metadata_size_on_device(dmesg_out) with TestRun.step("Dump the metadata of the cache"): dump_file_path = "/tmp/test_activate_corrupted.dump" md_dump = File(dump_file_path) md_dump.remove(force=True, ignore_errors=True) dd_count = int(md_size / Size(1, Unit.MebiByte)) + 1 ( Dd().input(active_cache_dev.path) .output(md_dump.full_path) .block_size(Size(1, Unit.MebiByte)) .count(dd_count) .run() ) md_dump.refresh_item() with TestRun.step("Stop cache instance."): active_cache.stop() with TestRun.step("Start standby cache instance."): standby_cache = casadm.standby_init(cache_dev=standby_cache_dev, cache_id=cache_id, cache_line_size=standby_cls, force=True) with TestRun.step("Verify if the cache exported object appeared in the system"): output = TestRun.executor.run_expect_success( f"ls -la /dev/ | grep {cache_exp_obj_name}" ) if output.stdout[0] != "b": TestRun.fail("The cache exported object is not a block device") with TestRun.step("Detach standby cache instance"): standby_cache.standby_detach() with TestRun.step(f"Copy changed metadata to the standby instance"): Dd().input(md_dump.full_path).output(standby_cache_dev.path).run() sync() with TestRun.step("Try to activate cache instance"): with pytest.raises(CmdException) as cmdExc: output = standby_cache.standby_activate(standby_cache_dev) if not check_stderr_msg(output, cache_line_size_mismatch): TestRun.LOGGER.error( f'Expected error message in format ' f'"{cache_line_size_mismatch[0]}"' f'Got "{output.stderr}" instead.' ) assert "Failed to activate standby cache." in str(cmdExc.value) with TestRun.step("Verify if cache is in standby detached state after failed activation"): cache_status = standby_cache.get_status() if cache_status != CacheStatus.standby_detached: TestRun.LOGGER.error( f'Expected Cache state: "{CacheStatus.standby.value}" ' f'Got "{cache_status.value}" instead.' ) @pytest.mark.CI @pytest.mark.skip(reason="Standby mode is not supported") @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane])) @pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) def test_standby_init_with_preexisting_metadata(): """ title: Initialize standby cache instance with preexisting metadata description: | Initialize standby cache instance with preexisting metadata with and without force flag pass_criteria: - initialize cache without force flag fails and informative error message is printed - initialize cache with force flag succeeds and passive instance is present in system """ with TestRun.step("Prepare device for cache"): cache_device = TestRun.disks["cache"] cache_device.create_partitions([Size(200, Unit.MebiByte)]) cache_device = cache_device.partitions[0] cls = CacheLineSize.LINE_32KiB cache_id = 1 with TestRun.step("Start standby cache instance"): cache = casadm.standby_init( cache_dev=cache_device, cache_line_size=cls, cache_id=cache_id, force=True, ) with TestRun.step("Stop standby cache instance"): cache.stop() with TestRun.step("Try initialize cache without force flag"): output = TestRun.executor.run( standby_init_cmd( cache_dev=cache_device.path, cache_id=str(cache_id), cache_line_size=str(int(cls.value.value / Unit.KibiByte.value)), ) ) if not check_stderr_msg(output, start_cache_with_existing_metadata): TestRun.LOGGER.error( f"Invalid error message. Expected {start_cache_with_existing_metadata}." f"Got {output.stderr}" ) with TestRun.step("Try initialize cache with force flag"): casadm.standby_init( cache_dev=cache_device, cache_line_size=cls, cache_id=cache_id, force=True, ) with TestRun.step("Verify whether passive instance is running"): caches = casadm_parser.get_caches() if len(caches) != 1: TestRun.LOGGER.error("Standby cache instance is not running!") @pytest.mark.CI @pytest.mark.skip(reason="Standby mode is not supported") @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.nand, DiskType.optane])) @pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) @pytest.mark.parametrizex("filesystem", Filesystem) def test_standby_init_with_preexisting_filesystem(filesystem): """ title: Initialize standby cache instance with preexisting filesystem description: | Initialize standby cache instance with preexisting filesystem with and without force flag pass_criteria: - initialize cache without force flag fails and informative error message is printed - initialize cache with force flag succeeds and passive instance is present in system """ with TestRun.step("Prepare device for cache"): cache_device = TestRun.disks["cache"] cache_device.create_partitions([Size(200, Unit.MebiByte)]) cache_device = cache_device.partitions[0] cls = CacheLineSize.LINE_32KiB cache_id = 1 with TestRun.step("Create filesystem on cache device partition"): cache_device.create_filesystem(filesystem) with TestRun.step("Try initialize cache without force flag"): output = TestRun.executor.run( standby_init_cmd( cache_dev=cache_device.path, cache_id=str(cache_id), cache_line_size=str(int(cls.value.value / Unit.KibiByte.value)), ) ) if not check_stderr_msg(output, standby_init_with_existing_filesystem): TestRun.LOGGER.error( f"Invalid error message. Expected {standby_init_with_existing_filesystem}." f"Got {output.stderr}" ) with TestRun.step("Try initialize cache with force flag"): casadm.standby_init( cache_dev=cache_device, cache_line_size=cls, cache_id=cache_id, force=True, ) with TestRun.step("Verify whether passive instance is running"): caches = casadm_parser.get_caches() if len(caches) != 1: TestRun.LOGGER.error("Standby cache instance is not running!") @pytest.mark.CI @pytest.mark.skip(reason="Standby mode is not supported") @pytest.mark.require_disk("caches", DiskTypeSet([DiskType.nand, DiskType.optane])) @pytest.mark.require_disk("core", DiskTypeLowerThan("caches")) def test_standby_activate_with_corepool(): """ title: Activate standby cache instance with corepool description: | Activation of standby cache with core taken from core pool pass_criteria: - During activate metadata on the device match with metadata in DRAM - Core is in active state after activate """ with TestRun.step("Prepare cache and core devices"): caches_dev = TestRun.disks["caches"] caches_dev.create_partitions([Size(500, Unit.MebiByte), Size(500, Unit.MebiByte)]) active_cache_dev = caches_dev.partitions[0] standby_cache_dev = caches_dev.partitions[1] core_dev = TestRun.disks["core"] core_dev.create_partitions([Size(200, Unit.MebiByte)]) core_dev = core_dev.partitions[0] cache_id = 1 core_id = 1 cache_exp_obj_name = f"cas-cache-{cache_id}" cls = CacheLineSize.LINE_16KiB with TestRun.step("Start regular cache instance"): cache = casadm.start_cache(cache_dev=active_cache_dev, cache_line_size=cls, cache_id=cache_id) with TestRun.step("Add core to regular cache instance"): cache.add_core(core_dev) with TestRun.step("Stop regular cache instance"): cache.stop() with TestRun.step("Add previously used core device to core pool using --try-add flag."): core = casadm.try_add(core_device=core_dev, cache_id=cache_id, core_id=core_id) with TestRun.step("Start standby cache instance."): standby_cache = casadm.standby_init(cache_dev=standby_cache_dev, cache_id=cache_id, cache_line_size=cls, force=True) with TestRun.step(f"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"): standby_cache.standby_detach() with TestRun.step("Activate standby cache instance"): standby_cache.standby_activate(standby_cache_dev) with TestRun.step("Check core status."): if core.get_status() is not CoreStatus.active: TestRun.fail(f"First core status should be active but is {core.get_status()}.") @pytest.mark.CI @pytest.mark.skip(reason="Standby mode is not supported") @pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) @pytest.mark.parametrizex("cache_line_size", CacheLineSize) def test_standby_start_stop(cache_line_size): """ title: Start and stop a standby cache instance. description: Test if cache can be started in standby state and stopped without activation. pass_criteria: - A cache exported object appears after starting a cache in standby state - The data written to the cache exported object committed on the underlying cache device - The cache exported object disappears after stopping the standby cache instance """ with TestRun.step("Prepare a cache device"): cache_size = Size(500, Unit.MebiByte) cache_device = TestRun.disks['cache'] cache_device.create_partitions([cache_size] * 2) cache_part = cache_device.partitions[0] with TestRun.step("Prepare partition with data to be written on the cache and count md5sum"): data_part = cache_device.partitions[1] block_size = Size(4, Unit.MebiByte) offset = int(cache_size / block_size / 2) Dd().output(data_part.path) \ .input("/dev/urandom") \ .block_size(block_size) \ .seek(offset) \ .run() data_md5sum_cmd = f"dd if={data_part.path} bs={block_size.value} skip={offset} | md5sum" data_md5 = TestRun.executor.run(data_md5sum_cmd).stdout.split()[0] with TestRun.step("Start standby cache with the tested cache line size"): cache_id = 1 cache = casadm.standby_init(cache_part, cache_id, cache_line_size, force=True) with TestRun.step("Verify that a cache exported object appeared in the system"): output = TestRun.executor.run_expect_success(f"ls -la /dev/ | grep 'cas-cache-{cache_id}'") if output.stdout[0] != "b": TestRun.fail("The cache exported object is not a block device") with TestRun.step("Sequentially write data to the cache exported" "object (with offset set to the middle of the device)"): Dd().output(f"/dev/cas-cache-{cache_id}") \ .input(data_part.path) \ .block_size(block_size) \ .seek(offset) \ .skip(offset) \ .run() with TestRun.step("Stop the cache instance"): cache.stop() with TestRun.step("Count checksum of data read from the cache device " "and compare with the checksum of data writen to the standby cache"): md5sum_cmd = f"dd if={cache_part.path} bs={block_size.value} skip={offset} | md5sum" cache_part_md5 = TestRun.executor.run(md5sum_cmd).stdout.split()[0] if data_md5 != cache_part_md5: TestRun.fail("The counted checksum of the data read from cache device" " and written to the standby cache differ")