diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/__init__.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/__init__.py new file mode 100644 index 0000000..18aaf34 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/__init__.py @@ -0,0 +1,4 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_cache_id.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_cache_id.py new file mode 100644 index 0000000..ad2719c --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_cache_id.py @@ -0,0 +1,86 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_cleaning_acp_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_cleaning_acp_cache_id( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for acp cleaning – cache id. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter' + command for acp cleaning policy parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + valid_values = [str(core.cache_id).encode("ascii")] + PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml")) + base_cmd = set_param_cleaning_acp_cmd(cache_id="{param}", wake_up="10") + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Cache id", + is_valid=cmd.param in valid_values, + ) diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_flush_max_buffers.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_flush_max_buffers.py new file mode 100644 index 0000000..e33768f --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_flush_max_buffers.py @@ -0,0 +1,99 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_cleaning_acp_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + +flush_max_buffers_min = 1 +flush_max_buffers_max = 10000 + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_cleaning_acp_flush_max_buffers( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for acp cleaning – flush max buffers. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong flush max buffers value + in 'set parameter' command for acp cleaning parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + PeachFuzzer.generate_config(get_fuzz_config("uint.yml")) + base_cmd = set_param_cleaning_acp_cmd( + cache_id=str(core.cache_id), + flush_max_buffers="{param}", + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Flush max buffers", + is_valid=__is_valid(cmd.param), + ) + + +def __is_valid(parameter): + try: + value = int(parameter) + except ValueError: + return False + return flush_max_buffers_min <= value <= flush_max_buffers_max diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_wake_up.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_wake_up.py new file mode 100644 index 0000000..c1b1f4b --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_wake_up.py @@ -0,0 +1,96 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_cleaning_acp_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + +wake_up_min = 0 +wake_up_max = 10000 + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_cleaning_acp_wake_up( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for acp cleaning – wake up. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong wake up value in + 'set parameter' command for acp cleaning parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + PeachFuzzer.generate_config(get_fuzz_config("uint.yml")) + base_cmd = set_param_cleaning_acp_cmd(cache_id=str(core.cache_id), wake_up="{param}") + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Wake up", + is_valid=__is_valid(cmd.param), + ) + + +def __is_valid(parameter): + try: + value = int(parameter) + except ValueError: + return False + return wake_up_min <= value <= wake_up_max diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_activity_threshold.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_activity_threshold.py new file mode 100644 index 0000000..50b0285 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_activity_threshold.py @@ -0,0 +1,99 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_cleaning_alru_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + +activity_threshold_min = 0 +activity_threshold_max = 1000000 + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_cleaning_alru_activity_threshold( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for alru cleaning – activity threshold. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong activity threshold value in + 'set parameter' command for alru cleaning parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + PeachFuzzer.generate_config(get_fuzz_config("uint.yml")) + base_cmd = set_param_cleaning_alru_cmd( + cache_id=str(core.cache_id), + activity_threshold="{param}", + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Activity threshold", + is_valid=__is_valid(cmd.param), + ) + + +def __is_valid(parameter): + try: + value = int(parameter) + except ValueError: + return False + return activity_threshold_min <= value <= activity_threshold_max diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_cache_id.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_cache_id.py new file mode 100644 index 0000000..c1d5399 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_cache_id.py @@ -0,0 +1,88 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_cleaning_alru_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + +default_wake_up = 20 + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_cleaning_alru_cache_id( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for alru cleaning – cache id. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter' + command for alru cleaning policy parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + valid_values = [str(core.cache_id).encode("ascii")] + PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml")) + base_cmd = set_param_cleaning_alru_cmd(cache_id="{param}", wake_up=str(default_wake_up)) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Cache id", + is_valid=cmd.param in valid_values, + ) diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_flush_max_buffers.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_flush_max_buffers.py new file mode 100644 index 0000000..94d19a5 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_flush_max_buffers.py @@ -0,0 +1,99 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_cleaning_alru_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + +flush_max_buffers_min = 1 +flush_max_buffers_max = 10000 + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_cleaning_alru_flush_max_buffers( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for alru cleaning – flush max buffers. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong flush max buffers value in + 'set parameter' command for alru cleaning parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + PeachFuzzer.generate_config(get_fuzz_config("uint.yml")) + base_cmd = set_param_cleaning_alru_cmd( + cache_id=str(core.cache_id), + flush_max_buffers="{param}", + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Flush max buffers", + is_valid=__is_valid(cmd.param), + ) + + +def __is_valid(parameter): + try: + value = int(parameter) + except ValueError: + return False + return flush_max_buffers_min <= value <= flush_max_buffers_max diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_staleness_time.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_staleness_time.py new file mode 100644 index 0000000..d315dbc --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_staleness_time.py @@ -0,0 +1,99 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_cleaning_alru_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + +staleness_time_min = 1 +staleness_time_max = 3600 + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_cleaning_alru_staleness_time( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for alru cleaning – staleness time. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong staleness time value in + 'set parameter' command for alru cleaning parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + PeachFuzzer.generate_config(get_fuzz_config("uint.yml")) + base_cmd = set_param_cleaning_alru_cmd( + cache_id=str(core.cache_id), + staleness_time="{param}", + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Staleness time", + is_valid=__is_valid(cmd.param), + ) + + +def __is_valid(parameter): + try: + value = int(parameter) + except ValueError: + return False + return staleness_time_min <= value <= staleness_time_max diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_wake_up.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_wake_up.py new file mode 100644 index 0000000..d6b6a0b --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_wake_up.py @@ -0,0 +1,96 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_cleaning_alru_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + +wake_up_min = 0 +wake_up_max = 3600 + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_cleaning_alru_wake_up( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for alru cleaning – wake up. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong wake up value in + 'set parameter' command for alru cleaning parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + PeachFuzzer.generate_config(get_fuzz_config("uint.yml")) + base_cmd = set_param_cleaning_alru_cmd(cache_id=str(core.cache_id), wake_up="{param}") + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Wake up", + is_valid=__is_valid(cmd.param.decode("ascii", "ignore")), + ) + + +def __is_valid(parameter): + try: + value = int(parameter) + except ValueError: + return False + return wake_up_min <= value <= wake_up_max diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_cache_id.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_cache_id.py new file mode 100644 index 0000000..471e930 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_cache_id.py @@ -0,0 +1,93 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, + CleaningPolicy, +) +from api.cas.cli import set_param_cleaning_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_cleaning_cache_id( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for cleaning – cache id. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter' + command for cleaning policy parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + valid_values = [str(core.cache_id).encode("ascii")] + PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml")) + base_cmd = set_param_cleaning_cmd( + cache_id="{param}", + policy=CleaningPolicy.nop.value.lower(), + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + output = run_cmd_and_validate( + cmd=cmd, + value_name="Cache id", + is_valid=cmd.param in valid_values, + ) + if output.exit_code == 0: + with TestRun.step("Set default cleaning policy."): + cache.set_cleaning_policy(cleaning_policy=CleaningPolicy.DEFAULT) diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_policy.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_policy.py new file mode 100644 index 0000000..87ebf32 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_policy.py @@ -0,0 +1,92 @@ +# +# Copyright(c) 2022 Intel Corporation +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheMode, + CacheLineSize, + CleaningPolicy, + UnalignedIo, + KernelParameters, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_cleaning_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + prepare_cas_instance, + get_fuzz_config, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("cleaning_policy", CleaningPolicy) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_cleaning_policy( + cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for cleaning - policy. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong cleaning policy in + 'set param cleaning' command. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step( + "Start cache with configuration and add core device, make filesystem and mount it" + ): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cleaning_policy=cleaning_policy, + mount_point=mount_point, + ) + + with TestRun.step("Run fio in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Prepare PeachFuzzer"): + valid_values = [e.name.encode("ascii").lower() for e in list(CleaningPolicy)] + PeachFuzzer.generate_config(get_fuzz_config("cleaning_policy.yml")) + base_cmd = set_param_cleaning_cmd(cache_id=str(cache.cache_id), policy="{param}") + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Policy", + is_valid=cmd.param in valid_values, + ) diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_cache_id.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_cache_id.py new file mode 100644 index 0000000..bc7adb5 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_cache_id.py @@ -0,0 +1,90 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, + PromotionPolicy, +) +from api.cas.cli import set_param_promotion_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_promotion_cache_id( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for promotion – cache id. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter' + command for promotion policy parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + valid_values = [str(core.cache_id).encode("ascii")] + PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml")) + base_cmd = set_param_promotion_cmd( + cache_id="{param}", + policy=PromotionPolicy.DEFAULT.value.lower(), + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Cache id", + is_valid=cmd.param in valid_values, + ) diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_cache_id.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_cache_id.py new file mode 100644 index 0000000..d53d5a3 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_cache_id.py @@ -0,0 +1,84 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_promotion_nhit_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_promotion_nhit_cache_id( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for promotion nhit – cache id. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter' + command for promotion nhit parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + valid_values = [str(core.cache_id).encode("ascii")] + PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml")) + base_cmd = set_param_promotion_nhit_cmd(cache_id="{param}", threshold="3") + commands = PeachFuzzer.get_fuzzed_command(base_cmd, TestRun.usr.fuzzy_iter_count) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Cache id", + is_valid=cmd.param in valid_values, + ) diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_threshold.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_threshold.py new file mode 100644 index 0000000..9854aa9 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_threshold.py @@ -0,0 +1,99 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_promotion_nhit_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + +threshold_min = 2 +threshold_max = 1000 + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_promotion_nhit_threshold( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for promotion nhit – threshold. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong threshold value in + 'set parameter' command for promotion nhit parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + PeachFuzzer.generate_config(get_fuzz_config("uint.yml")) + base_cmd = set_param_promotion_nhit_cmd( + cache_id=str(core.cache_id), + threshold="{param}", + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Threshold", + is_valid=__is_valid(cmd.param), + ) + + +def __is_valid(parameter): + try: + value = int(parameter) + except ValueError: + return False + return threshold_min <= value <= threshold_max diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_trigger.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_trigger.py new file mode 100644 index 0000000..21ae030 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_trigger.py @@ -0,0 +1,99 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_promotion_nhit_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + +trigger_min = 0 +trigger_max = 100 + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_promotion_nhit_trigger( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for promotion nhit – trigger. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong trigger value in + 'set parameter' command for promotion nhit parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + PeachFuzzer.generate_config(get_fuzz_config("uint.yml")) + base_cmd = set_param_promotion_nhit_cmd( + cache_id=str(core.cache_id), + trigger="{param}", + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Trigger", + is_valid=__is_valid(cmd.param), + ) + + +def __is_valid(parameter): + try: + value = int(parameter) + except ValueError: + return False + return trigger_min <= value <= trigger_max diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_policy.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_policy.py new file mode 100644 index 0000000..ef36012 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_policy.py @@ -0,0 +1,92 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheMode, + CacheLineSize, + CleaningPolicy, + UnalignedIo, + KernelParameters, + UseIoScheduler, + CacheModeTrait, + PromotionPolicy, +) +from api.cas.cli import set_param_promotion_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + prepare_cas_instance, + get_fuzz_config, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("cleaning_policy", CleaningPolicy) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_promotion_policy( + cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set param' command for promotion - policy. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong promotion policy in + 'set param promotion' command. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step( + "Start cache with configuration and add core device, make filesystem and mount it" + ): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cleaning_policy=cleaning_policy, + mount_point=mount_point, + ) + + with TestRun.step("Run fio in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Prepare PeachFuzzer"): + valid_values = [e.name.encode("ascii").lower() for e in list(PromotionPolicy)] + PeachFuzzer.generate_config(get_fuzz_config("promotion_policy.yml")) + base_cmd = set_param_promotion_cmd(cache_id=str(cache.cache_id), policy="{param}") + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Policy", + is_valid=cmd.param in valid_values, + ) diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_cache_id.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_cache_id.py new file mode 100644 index 0000000..efadf2c --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_cache_id.py @@ -0,0 +1,91 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, + SeqCutOffPolicy, +) +from api.cas.cli import set_param_cutoff_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_seq_cutoff_cache_id( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for sequential cutoff – cache id. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter' + command for sequential cutoff parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + valid_values = [str(core.cache_id).encode("ascii")] + PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml")) + base_cmd = set_param_cutoff_cmd( + cache_id="{param}", + core_id=str(core.core_id), + policy=SeqCutOffPolicy.DEFAULT.name.lower(), + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Cache id", + is_valid=cmd.param in valid_values, + ) diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_core_id.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_core_id.py new file mode 100644 index 0000000..63f6d27 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_core_id.py @@ -0,0 +1,91 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, + SeqCutOffPolicy, +) +from api.cas.cli import set_param_cutoff_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_seq_cutoff_core_id( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for sequential cutoff – core id. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong core id in 'set parameter' + command for sequential cutoff parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + valid_values = [str(core.core_id).encode("ascii")] + PeachFuzzer.generate_config(get_fuzz_config("core_id.yml")) + base_cmd = set_param_cutoff_cmd( + cache_id=str(core.cache_id), + core_id="{param}", + policy=SeqCutOffPolicy.DEFAULT.name.lower(), + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Core id", + is_valid=cmd.param in valid_values, + ) diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_policy.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_policy.py new file mode 100644 index 0000000..0de81ee --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_policy.py @@ -0,0 +1,97 @@ +# +# Copyright(c) 2022 Intel Corporation +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheMode, + CacheLineSize, + CleaningPolicy, + UnalignedIo, + KernelParameters, + UseIoScheduler, + CacheModeTrait, + SeqCutOffPolicy, +) +from api.cas.cli import set_param_cutoff_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + prepare_cas_instance, + get_fuzz_config, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("cleaning_policy", CleaningPolicy) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_seq_cutoff_policy( + cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set param' command for sequential cutoff - policy. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong sequential cutoff policy in + 'set param' command. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step( + "Start cache with configuration and add core device, make filesystem and " "mount it" + ): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cleaning_policy=cleaning_policy, + mount_point=mount_point, + ) + + with TestRun.step("Run fio in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Prepare PeachFuzzer"): + valid_values = [e.name.encode("ascii").lower() for e in list(SeqCutOffPolicy)] + PeachFuzzer.generate_config(get_fuzz_config("seq_cutoff_policy.yml")) + base_cmd = set_param_cutoff_cmd( + cache_id=str(cache.cache_id), + core_id=str(core.core_id), + policy="{param}", + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Policy", + is_valid=cmd.param in valid_values, + ) diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_promotion_count.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_promotion_count.py new file mode 100644 index 0000000..a19d791 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_promotion_count.py @@ -0,0 +1,101 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import numpy +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_cutoff_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + +promotion_count_min = 1 +promotion_count_max = numpy.iinfo(numpy.uint16).max + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_seq_cutoff_promotion_count( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for sequential cutoff – promotion count. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong promotion count in + 'set parameter' command for sequential cutoff parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + PeachFuzzer.generate_config(get_fuzz_config("uint.yml")) + base_cmd = set_param_cutoff_cmd( + cache_id=str(core.cache_id), + core_id=str(core.core_id), + promotion_count="{param}", + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Promotion count", + is_valid=__is_valid(cmd.param), + ) + + +def __is_valid(parameter): + try: + value = int(parameter) + except ValueError: + return False + return promotion_count_min <= value <= promotion_count_max diff --git a/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_threshold.py b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_threshold.py new file mode 100644 index 0000000..c7619c1 --- /dev/null +++ b/test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_threshold.py @@ -0,0 +1,100 @@ +# +# Copyright(c) 2024 Huawei Technologies Co., Ltd. +# SPDX-License-Identifier: BSD-3-Clause +# + +import pytest + +from api.cas.cache_config import ( + CacheLineSize, + CacheMode, + KernelParameters, + UnalignedIo, + UseIoScheduler, + CacheModeTrait, +) +from api.cas.cli import set_param_cutoff_cmd +from core.test_run import TestRun +from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet +from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer +from tests.security.fuzzy.kernel.common.common import ( + get_fuzz_config, + prepare_cas_instance, + run_cmd_and_validate, +) +from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import ( + get_basic_workload, + mount_point, +) + +threshold_min = 1 +threshold_max = 4194181 + + +@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand])) +@pytest.mark.require_disk("core", DiskTypeLowerThan("cache")) +@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites)) +@pytest.mark.parametrizex("cache_line_size", CacheLineSize) +@pytest.mark.parametrizex("unaligned_io", UnalignedIo) +@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler) +def test_fuzzy_set_param_seq_cutoff_threshold( + cache_mode, cache_line_size, unaligned_io, use_io_scheduler +): + """ + title: Fuzzy test for casadm 'set parameter' command for sequential cutoff – threshold. + description: | + Using Peach Fuzzer check Open CAS ability of handling wrong threshold in 'set parameter' + command for sequential cutoff parameters. + pass_criteria: + - System did not crash + - Open CAS still works. + """ + with TestRun.step("Start cache and add core device"): + cache_disk = TestRun.disks["cache"] + core_disk = TestRun.disks["core"] + cache, core = prepare_cas_instance( + kernel_params=KernelParameters(unaligned_io, use_io_scheduler), + cache_device=cache_disk, + core_device=core_disk, + cache_mode=cache_mode, + cache_line_size=cache_line_size, + mount_point=mount_point, + ) + + with TestRun.step("Run FIO in background"): + fio = get_basic_workload(mount_point) + fio_pid = fio.run_in_background() + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."): + PeachFuzzer.generate_config(get_fuzz_config("uint.yml")) + base_cmd = set_param_cutoff_cmd( + cache_id=str(core.cache_id), + core_id=str(core.core_id), + threshold="{param}", + ) + commands = PeachFuzzer.get_fuzzed_command( + command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count + ) + + for index, cmd in TestRun.iteration( + enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times" + ): + with TestRun.step(f"Iteration {index + 1}"): + if not TestRun.executor.check_if_process_exists(fio_pid): + raise Exception("Fio is not running.") + + run_cmd_and_validate( + cmd=cmd, + value_name="Threshold", + is_valid=__is_valid(cmd.param), + ) + + +def __is_valid(parameter): + try: + value = int(parameter) + except ValueError: + return False + return threshold_min <= value <= threshold_max