add fuzzy statistics and fuzzy io class tests

Signed-off-by: jwirkus <jakubx.wirkus@intel.com>

test/functional/api/cas/cli_messages.py

@@ -1,5 +1,5 @@
 #
-# Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2019-2022 Intel Corporation
 # SPDX-License-Identifier: BSD-3-Clause
 #
 import re
@@ -124,19 +124,23 @@ cache_dirty_shutdown = [
 ]
 
 
-def check_stderr_msg(output: Output, expected_messages):
-    return __check_string_msg(output.stderr, expected_messages)
+def check_stderr_msg(output: Output, expected_messages, negate=False):
+    return __check_string_msg(output.stderr, expected_messages, negate)
 
 
-def check_stdout_msg(output: Output, expected_messages):
-    return __check_string_msg(output.stdout, expected_messages)
+def check_stdout_msg(output: Output, expected_messages, negate=False):
+    return __check_string_msg(output.stdout, expected_messages, negate)
 
 
-def __check_string_msg(text: str, expected_messages):
+def __check_string_msg(text: str, expected_messages, negate=False):
     msg_ok = True
     for msg in expected_messages:
         matches = re.search(msg, text)
-        if not matches:
+        if not matches and not negate:
             TestRun.LOGGER.error(f"Message is incorrect, expected: {msg}\n actual: {text}.")
             msg_ok = False
+        elif matches and negate:
+            TestRun.LOGGER.error(f"Message is incorrect, expected to not find: {msg}\n "
+                                 f"actual: {text}.")
+            msg_ok = False
     return msg_ok

test/functional/tests/security/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#

test/functional/tests/security/fuzzy/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#

test/functional/tests/security/fuzzy/config/cache_id.yml

@@ -0,0 +1,11 @@
+- name: String
+  attributes:
+    name: CacheId
+    value: '1'
+    size: '14'
+    mutable: 'true'
+  children:
+   - name: Hint
+     attributes:
+       name: NumericalString
+       value: 'true'

test/functional/tests/security/fuzzy/config/core_id.yml

@@ -0,0 +1,11 @@
+- name: String
+  attributes:
+    name: CoreId
+    value: '1'
+    size: '12'
+    mutable: 'true'
+  children:
+   - name: Hint
+     attributes:
+       name: NumericalString
+       value: 'true'

test/functional/tests/security/fuzzy/config/filter.yml

@@ -0,0 +1,10 @@
+- name: String
+  attributes:
+    name: FilterValue
+    value: 'all'
+    mutable: 'true'
+  children:
+   - name: Hint
+     attributes:
+       name: ValidValues
+       value: 'conf;usage;req;blk;ext;'

test/functional/tests/security/fuzzy/config/io_class_id.yml

@@ -0,0 +1,11 @@
+- name: String
+  attributes:
+    name: IoClasssId
+    value: '0'
+    size: '8'
+    mutable: 'true'
+  children:
+   - name: Hint
+     attributes:
+       name: NumericalString
+       value: 'true'

test/functional/tests/security/fuzzy/config/output_format.yml

@@ -0,0 +1,10 @@
+- name: String
+  attributes:
+    name: Format
+    value: 'table'
+    mutable: 'true'
+  children:
+   - name: Hint
+     attributes:
+       name: ValidValues
+       value: 'csv;'

test/functional/tests/security/fuzzy/kernel/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#

test/functional/tests/security/fuzzy/kernel/common/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#

test/functional/tests/security/fuzzy/kernel/common/common.py

@@ -0,0 +1,64 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+import os
+import posixpath
+from typing import Callable
+
+import yaml
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, KernelParameters, CleaningPolicy
+from core.test_run import TestRun
+from test_tools.disk_utils import Filesystem
+from test_utils.size import Size, Unit
+
+
+def get_fuzz_config(config_name: str):
+    with open(posixpath.join(os.path.dirname(__file__), f"../../config/{config_name}")) as cfg:
+        fuzz_config = yaml.safe_load(cfg)
+
+    return fuzz_config
+
+
+def prepare_cas_instance(cache_disk, core_disk, cache_mode: CacheMode = None,
+                         cache_line_size: CacheLineSize = None,
+                         kernel_params: KernelParameters = KernelParameters(),
+                         cleaning_policy: CleaningPolicy = None, mount_point: str = None):
+    #  Change cleaning policy to default for Write Policy different than WB
+    if cleaning_policy:
+        cleaning_policy = CleaningPolicy.DEFAULT if cache_mode != CacheMode.WB \
+            else cleaning_policy
+
+    cache_disk.create_partitions([Size(400, Unit.MebiByte)])
+    cache_device = cache_disk.partitions[0]
+    cache = casadm.start_cache(cache_device, cache_mode, cache_line_size, 1, True,
+                               kernel_params=kernel_params)
+    if cleaning_policy:
+        cache.set_cleaning_policy(cleaning_policy)
+
+    if mount_point:
+        core_disk.create_filesystem(Filesystem.ext4)
+        core = cache.add_core(core_disk)
+        core.mount(mount_point)
+    else:
+        core = cache.add_core(core_disk)
+
+    return cache, core
+
+
+def run_cmd_and_validate(cmd, value_name: str, valid_values: list,
+                         post_process_param_func: Callable = None):
+    TestRun.LOGGER.info(f"{value_name}: {cmd.param}")
+    TestRun.LOGGER.info(f"Encoded command: {cmd.command}")
+    output = TestRun.executor.run(cmd.command)
+    param = cmd.param
+    if post_process_param_func:
+        param = post_process_param_func(param)
+
+    if output.exit_code == 0 and param not in valid_values:
+        TestRun.LOGGER.error(f" {param} value is not valid")
+    elif output.exit_code != 0 and param in valid_values:
+        TestRun.LOGGER.error(f" {param} value is valid but command returned with "
+                             f"{output.exit_code} exit code")

test/functional/tests/security/fuzzy/kernel/fuzzy_io_class/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#

test/functional/tests/security/fuzzy/kernel/fuzzy_io_class/test_fuzzy_io_class_list_cache_id.py

@@ -0,0 +1,59 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.casadm_params import OutputFormat
+from api.cas.cli import list_io_classes_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 get_fuzz_config, prepare_cas_instance, \
+    run_cmd_and_validate
+
+mount_point = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_io_class_list_cache_id(cache_mode, cache_line_size, cleaning_policy, unaligned_io,
+                                      use_io_scheduler):
+    """
+        title: Fuzzy test for casadm list IO class command – cache id
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong cache id in
+            ‘list IO class’ command.
+        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(cache_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy)
+
+    with TestRun.step("Load default IO class configuration file"):
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [str(core.cache_id).encode('ascii')]
+        PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml"))
+        base_cmd = list_io_classes_cmd("{param}", OutputFormat.table.name).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Cache_id", valid_values)

test/functional/tests/security/fuzzy/kernel/fuzzy_io_class/test_fuzzy_io_class_list_output_format.py

@@ -0,0 +1,59 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.casadm_params import OutputFormat
+from api.cas.cli import list_io_classes_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 get_fuzz_config, prepare_cas_instance, \
+    run_cmd_and_validate
+
+mount_point = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_io_class_list_output_format(cache_mode, cache_line_size, cleaning_policy,
+                                           unaligned_io, use_io_scheduler):
+    """
+        title: Fuzzy test for casadm list IO class command – output format
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong output format in
+            ‘list IO class’ command.
+        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(cache_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy)
+
+    with TestRun.step("Load default IO class configuration file"):
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [e.name.encode('ascii') for e in list(OutputFormat)]
+        PeachFuzzer.generate_config(get_fuzz_config("output_format.yml"))
+        base_cmd = list_io_classes_cmd(str(core.cache_id), "{param}").encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Output_format", valid_values)

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/common/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/common/common.py

@@ -0,0 +1,26 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+from datetime import timedelta
+
+from test_tools.fio.fio import Fio
+from test_tools.fio.fio_param import IoEngine, ReadWrite
+from test_utils.size import Size, Unit
+
+
+def get_basic_workload(mount_point: str):
+    file_min_size = Size(10, Unit.Byte).get_value()
+    file_max_size = Size(512, Unit.KiB).get_value()
+    fio = (Fio()
+           .create_command()
+           .io_engine(IoEngine.libaio)
+           .direct()
+           .run_time(timedelta(days=1))
+           .time_based()
+           .directory(mount_point)
+           .read_write(ReadWrite.randrw)
+           .nr_files(1000)
+           .file_size_range([(file_min_size, file_max_size)])
+           .num_jobs(32))
+    return fio

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/__init__.py

@@ -0,0 +1,4 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_cache_id.py

@@ -0,0 +1,65 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_cache_id(cache_mode, cache_line_size, cleaning_policy,
+                                         unaligned_io, use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - cache id
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [str(cache.cache_id).encode('ascii')]
+        PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml"))
+        base_cmd = print_statistics_cmd(cache_id="{param}", by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Cache_id", valid_values)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_core_id.py

@@ -0,0 +1,66 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_core_id(cache_mode, cache_line_size, cleaning_policy, unaligned_io,
+                                        use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - core id
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [str(core.core_id).encode('ascii')]
+        PeachFuzzer.generate_config(get_fuzz_config("core_id.yml"))
+        base_cmd = print_statistics_cmd(cache_id=str(core.cache_id), core_id="{param}",
+                                        by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Core_id", valid_values)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_filter_cache.py

@@ -0,0 +1,67 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.casadm_params import StatsFilter
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_filter_cache(cache_mode, cache_line_size, cleaning_policy,
+                                             unaligned_io, use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - filter for cache
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [e.name.encode('ascii') for e in list(StatsFilter)]
+        PeachFuzzer.generate_config(get_fuzz_config('filter.yml'))
+        base_cmd = print_statistics_cmd(cache_id=str(core.cache_id), filter="{param}",
+                                        by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Filter", valid_values)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_filter_cache_io_class.py

@@ -0,0 +1,68 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.casadm_params import StatsFilter
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_filter_cache_io_class(cache_mode, cache_line_size, cleaning_policy,
+                                                      unaligned_io, use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - filter for cache IO class
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [e.name.encode('ascii') for e in list(StatsFilter)]
+        PeachFuzzer.generate_config(get_fuzz_config("filter.yml"))
+        base_cmd = print_statistics_cmd(cache_id=str(core.cache_id), io_class_id="0",
+                                        per_io_class=True, filter="{param}",
+                                        by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Filter", valid_values)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_filter_core.py

@@ -0,0 +1,67 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.casadm_params import StatsFilter
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_filter_core(cache_mode, cache_line_size, cleaning_policy,
+                                            unaligned_io, use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - filter for core
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [e.name.encode('ascii') for e in list(StatsFilter)]
+        PeachFuzzer.generate_config(get_fuzz_config("filter.yml"))
+        base_cmd = print_statistics_cmd(cache_id=str(core.cache_id), core_id=str(core.core_id),
+                                        filter="{param}", by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Filter", valid_values)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_filter_core_io_class.py

@@ -0,0 +1,68 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.casadm_params import StatsFilter
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_filter_core_io_class(cache_mode, cache_line_size, cleaning_policy,
+                                                     unaligned_io, use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - filter for core IO class
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [e.name.encode('ascii') for e in list(StatsFilter)]
+        PeachFuzzer.generate_config(get_fuzz_config("filter.yml"))
+        base_cmd = print_statistics_cmd(cache_id=str(core.cache_id), core_id=str(core.core_id),
+                                        io_class_id="0", per_io_class=True, filter="{param}",
+                                        by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Filter", valid_values)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_io_class_id_for_cache.py

@@ -0,0 +1,74 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_io_class_id_for_cache(cache_mode, cache_line_size, cleaning_policy,
+                                                      unaligned_io, use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - IO class id for cache
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [b'', b'0', b'1'] + [str(x).encode('ascii') for x in range(11, 23)]
+        PeachFuzzer.generate_config(get_fuzz_config('io_class_id.yml'))
+        base_cmd = print_statistics_cmd(cache_id=str(core.cache_id), per_io_class=True,
+                                        io_class_id="{param}", by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Io_class_id", valid_values,
+                                 post_process_param_func=__strip_value)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)
+
+
+def __strip_value(param):
+    param = param.rstrip(b'\x00\x20\n\t')
+    param = b'0' if not len(param.rstrip(b'0')) else param  # treat '00' as '0'
+
+    return param

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_io_class_id_for_core.py

@@ -0,0 +1,75 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_io_class_id_for_core(cache_mode, cache_line_size, cleaning_policy,
+                                                     unaligned_io, use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - IO class id for core
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [b'', b'0', b'1'] + [str(x).encode('ascii') for x in range(11, 23)]
+        PeachFuzzer.generate_config(get_fuzz_config("io_class_id.yml"))
+        base_cmd = print_statistics_cmd(cache_id=str(core.cache_id), core_id=str(core.core_id),
+                                        per_io_class=True, io_class_id="{param}",
+                                        by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Io_class_id", valid_values,
+                                 post_process_param_func=__strip_value)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)
+
+
+def __strip_value(param):
+    param = param.rstrip(b'\x00\x20\n\t')
+    param = b'0' if not len(param.rstrip(b'0')) else param  # treat '00' as '0'
+
+    return param

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_output_format_for_cache.py

@@ -0,0 +1,68 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.casadm_params import OutputFormat
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_output_format_for_cache(cache_mode, cache_line_size,
+                                                        cleaning_policy, unaligned_io,
+                                                        use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - output format for cache
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [e.name.encode('ascii') for e in list(OutputFormat)]
+        PeachFuzzer.generate_config(get_fuzz_config('output_format.yml'))
+        base_cmd = print_statistics_cmd(cache_id=str(core.cache_id), output_format="{param}",
+                                        by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Output_format", valid_values)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_output_format_for_cache_io_class_id.py

@@ -0,0 +1,69 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.casadm_params import OutputFormat
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_output_format_for_cache_io_class_id(cache_mode, cache_line_size,
+                                                                    cleaning_policy, unaligned_io,
+                                                                    use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - output format for cache IO class
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [e.name.encode('ascii') for e in list(OutputFormat)]
+        PeachFuzzer.generate_config(get_fuzz_config('output_format.yml'))
+        base_cmd = print_statistics_cmd(cache_id=str(core.cache_id), io_class_id="0",
+                                        per_io_class=True, output_format="{param}",
+                                        by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Output_format", valid_values)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_output_format_for_core.py

@@ -0,0 +1,67 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.casadm_params import OutputFormat
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_output_format_for_core(cache_mode, cache_line_size, cleaning_policy,
+                                                       unaligned_io, use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - output format for core
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [e.name.encode('ascii') for e in list(OutputFormat)]
+        PeachFuzzer.generate_config(get_fuzz_config("output_format.yml"))
+        base_cmd = print_statistics_cmd(cache_id=str(core.cache_id), core_id=str(core.core_id),
+                                        output_format="{param}", by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Output_format", valid_values)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_output_format_for_core_io_class_id.py

@@ -0,0 +1,69 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from api.cas import casadm
+from api.cas.cache_config import CacheMode, CacheLineSize, CleaningPolicy, UnalignedIo, \
+    KernelParameters, UseIoScheduler
+from api.cas.casadm_params import OutputFormat
+from api.cas.cli import print_statistics_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 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 = "/mnt/test"
+iterations_count = 1000
+
+
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@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_print_statistics_output_format_for_core_io_class_id(cache_mode, cache_line_size,
+                                                                   cleaning_policy, unaligned_io,
+                                                                   use_io_scheduler):
+    """
+        title: Fuzzy test for casadm print statistics command - output format for core IO class
+        description: Using Peach Fuzzer check Open CAS ability of handling wrong CLI print
+            statistics 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_disk, core_disk, cache_mode, cache_line_size,
+                                           KernelParameters(unaligned_io, use_io_scheduler),
+                                           cleaning_policy, mount_point=mount_point)
+        casadm.load_io_classes(cache.cache_id, '/etc/opencas/ioclass-config.csv')
+
+    with TestRun.step("Run fio in background"):
+        fio = get_basic_workload(mount_point)
+        fio_pid = fio.run_in_background()
+
+    with TestRun.step("Prepare PeachFuzzer"):
+        valid_values = [e.name.encode('ascii') for e in list(OutputFormat)]
+        PeachFuzzer.generate_config(get_fuzz_config("output_format.yml"))
+        base_cmd = print_statistics_cmd(cache_id=str(core.cache_id), core_id=str(core.core_id),
+                                        io_class_id="0", per_io_class=True, output_format="{param}",
+                                        by_id_path=False).encode('ascii')
+        commands = PeachFuzzer.get_fuzzed_command(base_cmd, iterations_count)
+
+    for index, cmd in TestRun.iteration(enumerate(commands), f"Run command {iterations_count} "
+                                                             f"times"):
+        with TestRun.step(f"Iteration {index + 1}"):
+            run_cmd_and_validate(cmd, "Output_format", valid_values)
+
+    with TestRun.step("Stop 'fio'"):
+        TestRun.executor.kill_process(fio_pid)