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

Fuzzy tests for casadm 'set param' command

Browse Source 
Signed-off-by: Katarzyna Treder <katarzyna.treder@h-partners.com>
This commit is contained in:
Katarzyna Treder 2024-08-22 14:43:30 +02:00 committed by Katarzyna Treder
parent 4efdf36f45
commit e8fbc624ed
21 changed files with 1895 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/__init__.py Normal file
View File

@ -0,0 +1,4 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#

86
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_cache_id.py Normal file
View File

@ -0,0 +1,86 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_cleaning_acp_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_cleaning_acp_cache_id(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for acp cleaning – cache id.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter'
command for acp cleaning policy parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
valid_values = [str(core.cache_id).encode("ascii")]
PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml"))
base_cmd = set_param_cleaning_acp_cmd(cache_id="{param}", wake_up="10")
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Cache id",
is_valid=cmd.param in valid_values,
)

99
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_flush_max_buffers.py Normal file
View File

@ -0,0 +1,99 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_cleaning_acp_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
flush_max_buffers_min = 1
flush_max_buffers_max = 10000
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_cleaning_acp_flush_max_buffers(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for acp cleaning – flush max buffers.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong flush max buffers value
in 'set parameter' command for acp cleaning parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
PeachFuzzer.generate_config(get_fuzz_config("uint.yml"))
base_cmd = set_param_cleaning_acp_cmd(
cache_id=str(core.cache_id),
flush_max_buffers="{param}",
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Flush max buffers",
is_valid=__is_valid(cmd.param),
)
def __is_valid(parameter):
try:
value = int(parameter)
except ValueError:
return False
return flush_max_buffers_min <= value <= flush_max_buffers_max

96
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_acp_wake_up.py Normal file
View File

@ -0,0 +1,96 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_cleaning_acp_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
wake_up_min = 0
wake_up_max = 10000
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_cleaning_acp_wake_up(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for acp cleaning – wake up.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong wake up value in
'set parameter' command for acp cleaning parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
PeachFuzzer.generate_config(get_fuzz_config("uint.yml"))
base_cmd = set_param_cleaning_acp_cmd(cache_id=str(core.cache_id), wake_up="{param}")
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Wake up",
is_valid=__is_valid(cmd.param),
)
def __is_valid(parameter):
try:
value = int(parameter)
except ValueError:
return False
return wake_up_min <= value <= wake_up_max

99
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_activity_threshold.py Normal file
View File

@ -0,0 +1,99 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_cleaning_alru_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
activity_threshold_min = 0
activity_threshold_max = 1000000
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_cleaning_alru_activity_threshold(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for alru cleaning – activity threshold.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong activity threshold value in
'set parameter' command for alru cleaning parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
PeachFuzzer.generate_config(get_fuzz_config("uint.yml"))
base_cmd = set_param_cleaning_alru_cmd(
cache_id=str(core.cache_id),
activity_threshold="{param}",
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Activity threshold",
is_valid=__is_valid(cmd.param),
)
def __is_valid(parameter):
try:
value = int(parameter)
except ValueError:
return False
return activity_threshold_min <= value <= activity_threshold_max

88
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_cache_id.py Normal file
View File

@ -0,0 +1,88 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_cleaning_alru_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
default_wake_up = 20
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_cleaning_alru_cache_id(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for alru cleaning – cache id.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter'
command for alru cleaning policy parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
valid_values = [str(core.cache_id).encode("ascii")]
PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml"))
base_cmd = set_param_cleaning_alru_cmd(cache_id="{param}", wake_up=str(default_wake_up))
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Cache id",
is_valid=cmd.param in valid_values,
)

99
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_flush_max_buffers.py Normal file
View File

@ -0,0 +1,99 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_cleaning_alru_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
flush_max_buffers_min = 1
flush_max_buffers_max = 10000
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_cleaning_alru_flush_max_buffers(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for alru cleaning – flush max buffers.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong flush max buffers value in
'set parameter' command for alru cleaning parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
PeachFuzzer.generate_config(get_fuzz_config("uint.yml"))
base_cmd = set_param_cleaning_alru_cmd(
cache_id=str(core.cache_id),
flush_max_buffers="{param}",
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Flush max buffers",
is_valid=__is_valid(cmd.param),
)
def __is_valid(parameter):
try:
value = int(parameter)
except ValueError:
return False
return flush_max_buffers_min <= value <= flush_max_buffers_max

99
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_staleness_time.py Normal file
View File

@ -0,0 +1,99 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_cleaning_alru_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
staleness_time_min = 1
staleness_time_max = 3600
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_cleaning_alru_staleness_time(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for alru cleaning – staleness time.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong staleness time value in
'set parameter' command for alru cleaning parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
PeachFuzzer.generate_config(get_fuzz_config("uint.yml"))
base_cmd = set_param_cleaning_alru_cmd(
cache_id=str(core.cache_id),
staleness_time="{param}",
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Staleness time",
is_valid=__is_valid(cmd.param),
)
def __is_valid(parameter):
try:
value = int(parameter)
except ValueError:
return False
return staleness_time_min <= value <= staleness_time_max

96
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_alru_wake_up.py Normal file
View File

@ -0,0 +1,96 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_cleaning_alru_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
wake_up_min = 0
wake_up_max = 3600
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_cleaning_alru_wake_up(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for alru cleaning – wake up.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong wake up value in
'set parameter' command for alru cleaning parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
PeachFuzzer.generate_config(get_fuzz_config("uint.yml"))
base_cmd = set_param_cleaning_alru_cmd(cache_id=str(core.cache_id), wake_up="{param}")
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Wake up",
is_valid=__is_valid(cmd.param.decode("ascii", "ignore")),
)
def __is_valid(parameter):
try:
value = int(parameter)
except ValueError:
return False
return wake_up_min <= value <= wake_up_max

93
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_cache_id.py Normal file
View File

@ -0,0 +1,93 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
CleaningPolicy,
)
from api.cas.cli import set_param_cleaning_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_cleaning_cache_id(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for cleaning – cache id.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter'
command for cleaning policy parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
valid_values = [str(core.cache_id).encode("ascii")]
PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml"))
base_cmd = set_param_cleaning_cmd(
cache_id="{param}",
policy=CleaningPolicy.nop.value.lower(),
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
output = run_cmd_and_validate(
cmd=cmd,
value_name="Cache id",
is_valid=cmd.param in valid_values,
)
if output.exit_code == 0:
with TestRun.step("Set default cleaning policy."):
cache.set_cleaning_policy(cleaning_policy=CleaningPolicy.DEFAULT)

92
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_cleaning_policy.py Normal file
View File

@ -0,0 +1,92 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheMode,
CacheLineSize,
CleaningPolicy,
UnalignedIo,
KernelParameters,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_cleaning_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
prepare_cas_instance,
get_fuzz_config,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("cleaning_policy", CleaningPolicy)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_cleaning_policy(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for cleaning - policy.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong cleaning policy in
'set param cleaning' command.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step(
"Start cache with configuration and add core device, make filesystem and mount it"
):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cleaning_policy=cleaning_policy,
mount_point=mount_point,
)
with TestRun.step("Run fio in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Prepare PeachFuzzer"):
valid_values = [e.name.encode("ascii").lower() for e in list(CleaningPolicy)]
PeachFuzzer.generate_config(get_fuzz_config("cleaning_policy.yml"))
base_cmd = set_param_cleaning_cmd(cache_id=str(cache.cache_id), policy="{param}")
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Policy",
is_valid=cmd.param in valid_values,
)

90
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_cache_id.py Normal file
View File

@ -0,0 +1,90 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
PromotionPolicy,
)
from api.cas.cli import set_param_promotion_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_promotion_cache_id(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for promotion – cache id.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter'
command for promotion policy parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
valid_values = [str(core.cache_id).encode("ascii")]
PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml"))
base_cmd = set_param_promotion_cmd(
cache_id="{param}",
policy=PromotionPolicy.DEFAULT.value.lower(),
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Cache id",
is_valid=cmd.param in valid_values,
)

84
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_cache_id.py Normal file
View File

@ -0,0 +1,84 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_promotion_nhit_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_promotion_nhit_cache_id(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for promotion nhit – cache id.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter'
command for promotion nhit parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
valid_values = [str(core.cache_id).encode("ascii")]
PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml"))
base_cmd = set_param_promotion_nhit_cmd(cache_id="{param}", threshold="3")
commands = PeachFuzzer.get_fuzzed_command(base_cmd, TestRun.usr.fuzzy_iter_count)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Cache id",
is_valid=cmd.param in valid_values,
)

99
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_threshold.py Normal file
View File

@ -0,0 +1,99 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_promotion_nhit_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
threshold_min = 2
threshold_max = 1000
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_promotion_nhit_threshold(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for promotion nhit – threshold.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong threshold value in
'set parameter' command for promotion nhit parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
PeachFuzzer.generate_config(get_fuzz_config("uint.yml"))
base_cmd = set_param_promotion_nhit_cmd(
cache_id=str(core.cache_id),
threshold="{param}",
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Threshold",
is_valid=__is_valid(cmd.param),
)
def __is_valid(parameter):
try:
value = int(parameter)
except ValueError:
return False
return threshold_min <= value <= threshold_max

99
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_nhit_trigger.py Normal file
View File

@ -0,0 +1,99 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_promotion_nhit_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
trigger_min = 0
trigger_max = 100
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_promotion_nhit_trigger(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for promotion nhit – trigger.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong trigger value in
'set parameter' command for promotion nhit parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
PeachFuzzer.generate_config(get_fuzz_config("uint.yml"))
base_cmd = set_param_promotion_nhit_cmd(
cache_id=str(core.cache_id),
trigger="{param}",
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Trigger",
is_valid=__is_valid(cmd.param),
)
def __is_valid(parameter):
try:
value = int(parameter)
except ValueError:
return False
return trigger_min <= value <= trigger_max

92
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_promotion_policy.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.cache_config import (
CacheMode,
CacheLineSize,
CleaningPolicy,
UnalignedIo,
KernelParameters,
UseIoScheduler,
CacheModeTrait,
PromotionPolicy,
)
from api.cas.cli import set_param_promotion_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
prepare_cas_instance,
get_fuzz_config,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("cleaning_policy", CleaningPolicy)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_promotion_policy(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set param' command for promotion - policy.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong promotion policy in
'set param promotion' command.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step(
"Start cache with configuration and add core device, make filesystem and mount it"
):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cleaning_policy=cleaning_policy,
mount_point=mount_point,
)
with TestRun.step("Run fio in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Prepare PeachFuzzer"):
valid_values = [e.name.encode("ascii").lower() for e in list(PromotionPolicy)]
PeachFuzzer.generate_config(get_fuzz_config("promotion_policy.yml"))
base_cmd = set_param_promotion_cmd(cache_id=str(cache.cache_id), policy="{param}")
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Policy",
is_valid=cmd.param in valid_values,
)

91
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_cache_id.py Normal file
View File

@ -0,0 +1,91 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
SeqCutOffPolicy,
)
from api.cas.cli import set_param_cutoff_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_seq_cutoff_cache_id(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for sequential cutoff – cache id.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong cache id in 'set parameter'
command for sequential cutoff parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
valid_values = [str(core.cache_id).encode("ascii")]
PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml"))
base_cmd = set_param_cutoff_cmd(
cache_id="{param}",
core_id=str(core.core_id),
policy=SeqCutOffPolicy.DEFAULT.name.lower(),
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Cache id",
is_valid=cmd.param in valid_values,
)

91
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_core_id.py Normal file
View File

@ -0,0 +1,91 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
SeqCutOffPolicy,
)
from api.cas.cli import set_param_cutoff_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_seq_cutoff_core_id(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for sequential cutoff – core id.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong core id in 'set parameter'
command for sequential cutoff parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
valid_values = [str(core.core_id).encode("ascii")]
PeachFuzzer.generate_config(get_fuzz_config("core_id.yml"))
base_cmd = set_param_cutoff_cmd(
cache_id=str(core.cache_id),
core_id="{param}",
policy=SeqCutOffPolicy.DEFAULT.name.lower(),
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Core id",
is_valid=cmd.param in valid_values,
)

97
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_policy.py Normal file
View File

@ -0,0 +1,97 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheMode,
CacheLineSize,
CleaningPolicy,
UnalignedIo,
KernelParameters,
UseIoScheduler,
CacheModeTrait,
SeqCutOffPolicy,
)
from api.cas.cli import set_param_cutoff_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
prepare_cas_instance,
get_fuzz_config,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("cleaning_policy", CleaningPolicy)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_seq_cutoff_policy(
cache_mode, cache_line_size, cleaning_policy, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set param' command for sequential cutoff - policy.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong sequential cutoff policy in
'set param' command.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step(
"Start cache with configuration and add core device, make filesystem and " "mount it"
):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cleaning_policy=cleaning_policy,
mount_point=mount_point,
)
with TestRun.step("Run fio in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Prepare PeachFuzzer"):
valid_values = [e.name.encode("ascii").lower() for e in list(SeqCutOffPolicy)]
PeachFuzzer.generate_config(get_fuzz_config("seq_cutoff_policy.yml"))
base_cmd = set_param_cutoff_cmd(
cache_id=str(cache.cache_id),
core_id=str(core.core_id),
policy="{param}",
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Policy",
is_valid=cmd.param in valid_values,
)

101
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_promotion_count.py Normal file
View File

@ -0,0 +1,101 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import numpy
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_cutoff_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
promotion_count_min = 1
promotion_count_max = numpy.iinfo(numpy.uint16).max
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_seq_cutoff_promotion_count(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for sequential cutoff – promotion count.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong promotion count in
'set parameter' command for sequential cutoff parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
PeachFuzzer.generate_config(get_fuzz_config("uint.yml"))
base_cmd = set_param_cutoff_cmd(
cache_id=str(core.cache_id),
core_id=str(core.core_id),
promotion_count="{param}",
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Promotion count",
is_valid=__is_valid(cmd.param),
)
def __is_valid(parameter):
try:
value = int(parameter)
except ValueError:
return False
return promotion_count_min <= value <= promotion_count_max

100
test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_set_param/test_fuzzy_set_param_seq_cutoff_threshold.py Normal file
View File

@ -0,0 +1,100 @@
#
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
import pytest
from api.cas.cache_config import (
CacheLineSize,
CacheMode,
KernelParameters,
UnalignedIo,
UseIoScheduler,
CacheModeTrait,
)
from api.cas.cli import set_param_cutoff_cmd
from core.test_run import TestRun
from storage_devices.disk import DiskType, DiskTypeLowerThan, DiskTypeSet
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
from tests.security.fuzzy.kernel.common.common import (
get_fuzz_config,
prepare_cas_instance,
run_cmd_and_validate,
)
from tests.security.fuzzy.kernel.fuzzy_with_io.common.common import (
get_basic_workload,
mount_point,
)
threshold_min = 1
threshold_max = 4194181
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@pytest.mark.require_disk("core", DiskTypeLowerThan("cache"))
@pytest.mark.parametrizex("cache_mode", CacheMode.with_traits(CacheModeTrait.LazyWrites))
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
def test_fuzzy_set_param_seq_cutoff_threshold(
cache_mode, cache_line_size, unaligned_io, use_io_scheduler
):
"""
title: Fuzzy test for casadm 'set parameter' command for sequential cutoff – threshold.
description: |
Using Peach Fuzzer check Open CAS ability of handling wrong threshold in 'set parameter'
command for sequential cutoff parameters.
pass_criteria:
- System did not crash
- Open CAS still works.
"""
with TestRun.step("Start cache and add core device"):
cache_disk = TestRun.disks["cache"]
core_disk = TestRun.disks["core"]
cache, core = prepare_cas_instance(
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
cache_device=cache_disk,
core_device=core_disk,
cache_mode=cache_mode,
cache_line_size=cache_line_size,
mount_point=mount_point,
)
with TestRun.step("Run FIO in background"):
fio = get_basic_workload(mount_point)
fio_pid = fio.run_in_background()
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
with TestRun.step("Using Peach Fuzzer create 'set parameter' command and run it."):
PeachFuzzer.generate_config(get_fuzz_config("uint.yml"))
base_cmd = set_param_cutoff_cmd(
cache_id=str(core.cache_id),
core_id=str(core.core_id),
threshold="{param}",
)
commands = PeachFuzzer.get_fuzzed_command(
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
)
for index, cmd in TestRun.iteration(
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
):
with TestRun.step(f"Iteration {index + 1}"):
if not TestRun.executor.check_if_process_exists(fio_pid):
raise Exception("Fio is not running.")
run_cmd_and_validate(
cmd=cmd,
value_name="Threshold",
is_valid=__is_valid(cmd.param),
)
def __is_valid(parameter):
try:
value = int(parameter)
except ValueError:
return False
return threshold_min <= value <= threshold_max