github/open-cas-linux
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fuzzy tests for casadm 'print statistics' command

Browse Source 
Signed-off-by: Katarzyna Treder <katarzyna.treder@h-partners.com>
This commit is contained in:
Katarzyna Treder 2024-08-22 14:44:06 +02:00 committed by Katarzyna Treder
parent e8fbc624ed
commit b15076a22e
25 changed files with 1403 additions and 745 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_cache_filter.py Normal file
View File

@ -0,0 +1,96 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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,
)
@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_filter(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'print statistics' command for cache - filter
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong filter for cache in casadm
'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_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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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") 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
)
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="Filter",
is_valid=cmd.param in valid_values,
)

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

@ -1,22 +1,33 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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.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"
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]))
@ -26,39 +37,57 @@ mount_point = "/mnt/test"
@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):
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.
title: Fuzzy test for casadm 'print statistics' command - cache id
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong cache id in casadm
'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(
"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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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 = [str(cache.cache_id).encode('ascii')]
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, TestRun.usr.fuzzy_iter_count)
base_cmd = print_statistics_cmd(cache_id="{param}", by_id_path=False)
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"):
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
run_cmd_and_validate(cmd, "Cache_id", cmd.param in valid_values)
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)
run_cmd_and_validate(
cmd=cmd,
value_name="Cache id",
is_valid=cmd.param in valid_values,
)

99
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_cache_io_class_filter.py Normal file
View File

@ -0,0 +1,99 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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,
)
@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_io_class_filter(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'print statistics' command for cache IO class - filter
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong filter for cache IO class
in casadm '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_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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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") 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",
filter="{param}",
by_id_path=False,
)
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="Filter",
is_valid=cmd.param in valid_values,
)

104
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_cache_io_class_id.py Normal file
View File

@ -0,0 +1,104 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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,
)
@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_io_class_id(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'print statistics' command for cache - IO class id
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong IO class id for cache in
casadm '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_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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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 = [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),
io_class_id="{param}",
by_id_path=False,
)
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="Io class id",
is_valid=__is_valid(cmd.param, valid_values),
)
def __is_valid(param, valid_values):
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 in valid_values

99
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_cache_io_class_output_format.py Normal file
View File

@ -0,0 +1,99 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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,
)
@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_io_class_output_format(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'print statistics' command for cache IO class - output format
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong output format for
cache IO class in casadm '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_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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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") 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",
output_format="{param}",
by_id_path=False,
)
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="Output format",
is_valid=cmd.param in valid_values,
)

98
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_cache_output_format.py Normal file
View File

@ -0,0 +1,98 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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,
)
@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_output_format(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'print statistics' command for cache - output format
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong output format for cache in
casadm '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_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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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") 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,
)
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="Output format",
is_valid=cmd.param in valid_values,
)

99
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_core_filter.py Normal file
View File

@ -0,0 +1,99 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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,
)
@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_filter(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'print statistics' command for core - filter
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong filter for core in casadm
'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_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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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") 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,
)
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="Filter",
is_valid=cmd.param in valid_values,
)

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

@ -1,22 +1,33 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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.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"
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]))
@ -26,40 +37,61 @@ mount_point = "/mnt/test"
@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):
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.
title: Fuzzy test for casadm 'print statistics' command - core id
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong core id in casadm
'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(
"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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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 = [str(core.core_id).encode('ascii')]
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, TestRun.usr.fuzzy_iter_count)
base_cmd = print_statistics_cmd(
cache_id=str(core.cache_id),
core_id="{param}",
by_id_path=False,
)
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"):
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
run_cmd_and_validate(cmd, "Core_id", cmd.param in valid_values)
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)
run_cmd_and_validate(
cmd=cmd,
value_name="Core id",
is_valid=cmd.param in valid_values,
)

100
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_core_io_class_filter.py Normal file
View File

@ -0,0 +1,100 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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,
)
@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_io_class_filter(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'print statistics' command for core IO class - filter
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong filter for core IO class in
casadm '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_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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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") 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",
filter="{param}",
by_id_path=False,
)
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="Filter",
is_valid=cmd.param in valid_values,
)

105
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_core_io_class_id.py Normal file
View File

@ -0,0 +1,105 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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,
)
@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_io_class_id(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'print statistics' command for core - IO class id
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong IO class id for core
in casadm '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_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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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 = [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),
io_class_id="{param}",
by_id_path=False,
)
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="Io class id",
is_valid=__is_valid(cmd.param, valid_values),
)
def __is_valid(param, valid_values):
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 in valid_values

100
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_core_io_class_output_format.py Normal file
View File

@ -0,0 +1,100 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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,
)
@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_io_class_output_format(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'print statistics' command for core IO class - output format
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong output format for core
IO class in casadm '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_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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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") 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",
output_format="{param}",
by_id_path=False,
)
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="Output format",
is_valid=cmd.param in valid_values,
)

98
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_core_output_format.py Normal file
View File

@ -0,0 +1,98 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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,
)
@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_output_format(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'print statistics' command for core - output format
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong output format for core
in casadm '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_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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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") 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,
)
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="Output format",
is_valid=cmd.param in valid_values,
)

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

@ -1,66 +0,0 @@
#
# 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"
@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, 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}"):
run_cmd_and_validate(cmd, "Filter", cmd.param in valid_values)
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)

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

@ -1,67 +0,0 @@
#
# 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"
@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, 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}"):
run_cmd_and_validate(cmd, "Filter", cmd.param in valid_values)
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)

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

@ -1,66 +0,0 @@
#
# 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"
@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, 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}"):
run_cmd_and_validate(cmd, "Filter", cmd.param in valid_values)
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)

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

@ -1,67 +0,0 @@
#
# 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"
@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, 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}"):
run_cmd_and_validate(cmd, "Filter", cmd.param in valid_values)
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)

96
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_print_statistics_flag.py Normal file
View File

@ -0,0 +1,96 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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,
)
@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_flag(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'print statistics' command - flag
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong flag in casadm
'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_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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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 = ["", "-b", "--by-id-path"]
valid_values = [v.encode("ascii") for v in valid_values]
PeachFuzzer.generate_config(get_fuzz_config("flags.yml"))
base_cmd = (
print_statistics_cmd(cache_id=str(cache.cache_id), by_id_path=False) + " {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="Flag",
is_valid=cmd.param in valid_values,
)

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

@ -1,73 +0,0 @@
#
# 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"
@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, 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}"):
run_cmd_and_validate(cmd, "Io_class_id",
__is_valid_io_class_id(cmd.param, valid_values))
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)
def __is_valid_io_class_id(param, valid_values):
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 in valid_values

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

@ -1,74 +0,0 @@
#
# 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"
@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, 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}"):
run_cmd_and_validate(cmd, "Io_class_id",
__is_valid_io_class_id(cmd.param, valid_values))
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)
def __is_valid_io_class_id(param, valid_values):
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 in valid_values

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

@ -1,67 +0,0 @@
#
# 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"
@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, 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}"):
run_cmd_and_validate(cmd, "Output_format", cmd.param in valid_values)
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)

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

@ -1,68 +0,0 @@
#
# 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"
@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, 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}"):
run_cmd_and_validate(cmd, "Output_format", cmd.param in valid_values)
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)

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

@ -1,66 +0,0 @@
#
# 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"
@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, 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}"):
run_cmd_and_validate(cmd, "Output_format", cmd.param in valid_values)
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)

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

@ -1,68 +0,0 @@
#
# 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"
@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, 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}"):
run_cmd_and_validate(cmd, "Output_format", cmd.param in valid_values)
with TestRun.step("Stop 'fio'"):
TestRun.executor.kill_process(fio_pid)

93
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_reset_counters_cache_id.py Normal file
View File

@ -0,0 +1,93 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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 reset_counters_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)
@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_reset_counters_cache_id(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'reset counters' command - cache id.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong cache id in casadm
'reset counters' 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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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 = [str(cache.cache_id).encode("ascii")]
PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml"))
base_cmd = reset_counters_cmd(cache_id="{param}", core_id=str(core.core_id))
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,
)

92
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_statistics/test_fuzzy_reset_counters_core_id.py Normal file
View File

@ -0,0 +1,92 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# 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 reset_counters_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)
@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_reset_counters_core_id(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'reset counters' command - core id.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong core id in casadm
'reset counters' 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,
)
casadm.load_io_classes(cache_id=cache.cache_id, file="/etc/opencas/ioclass-config.csv")
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 = [str(core.core_id).encode("ascii")]
PeachFuzzer.generate_config(get_fuzz_config("core_id.yml"))
base_cmd = reset_counters_cmd(cache_id=str(core.cache_id), core_id="{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="Core id",
is_valid=cmd.param in valid_values,
)