Skip to content

Commit e9ea92d

Browse files
author
Katarzyna Treder
committed
Add fuzzy tests for attach/detach commands
Signed-off-by: Katarzyna Treder <katarzyna.treder@h-partners.com>
1 parent 3db09b2 commit e9ea92d

File tree

5 files changed

+366
-0
lines changed

5 files changed

+366
-0
lines changed

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

Whitespace-only changes.
Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
#
2+
# Copyright(c) 2025 Huawei Technologies Co., Ltd.
3+
# SPDX-License-Identifier: BSD-3-Clause
4+
#
5+
6+
import pytest
7+
8+
from api.cas import casadm
9+
from api.cas.cache_config import (
10+
CacheMode,
11+
CacheLineSize,
12+
UnalignedIo,
13+
KernelParameters,
14+
UseIoScheduler,
15+
)
16+
from api.cas.cli import attach_cache_cmd
17+
from core.test_run import TestRun
18+
from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
19+
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
20+
from type_def.size import Unit, Size
21+
from tests.security.fuzzy.kernel.common.common import (
22+
run_cmd_and_validate,
23+
get_device_fuzz_config,
24+
)
25+
26+
27+
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
28+
@pytest.mark.require_disk("other", DiskTypeLowerThan("cache"))
29+
@pytest.mark.parametrizex("cache_mode", CacheMode)
30+
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
31+
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
32+
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
33+
def test_fuzzy_attach_cache_cache_device(cache_mode, cache_line_size, unaligned_io, use_io_scheduler):
34+
"""
35+
title: Fuzzy test for casadm 'attach cache' command – cache device
36+
description: |
37+
Using Peach Fuzzer check ability of handling wrong device in 'attach cache' command.
38+
pass_criteria:
39+
- System did not crash
40+
"""
41+
42+
cache_id = 1
43+
44+
with TestRun.step("Create partitions on all devices"):
45+
available_disks = [TestRun.disks["cache"], TestRun.disks["other"]]
46+
for disk in available_disks:
47+
disk.create_partitions([Size(400, Unit.MebiByte)])
48+
49+
with TestRun.step("Start cache"):
50+
cache_disk = TestRun.disks["cache"]
51+
cache = casadm.start_cache(
52+
cache_dev=cache_disk.partitions[0],
53+
cache_mode=cache_mode,
54+
cache_line_size=cache_line_size,
55+
cache_id=cache_id,
56+
force=True,
57+
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
58+
)
59+
60+
with TestRun.step("Detach cache"):
61+
cache.detach()
62+
63+
with TestRun.step("Prepare PeachFuzzer"):
64+
disks_paths = [disk.path for disk in available_disks]
65+
partitions_paths = [disk.partitions[0].path for disk in available_disks]
66+
valid_values = disks_paths + partitions_paths
67+
# fuzz only partitions to speed up test
68+
fuzz_config = get_device_fuzz_config(partitions_paths)
69+
# forbidden values are created to prevent attaching other disks connected to DUT
70+
forbidden_values = [
71+
disk.path for disk in TestRun.dut.disks if disk.path not in valid_values
72+
]
73+
valid_values = [path.encode("ascii") for path in valid_values]
74+
forbidden_values = [path.encode("ascii") for path in forbidden_values]
75+
76+
PeachFuzzer.generate_config(fuzz_config)
77+
base_cmd = attach_cache_cmd(
78+
cache_dev="{param}",
79+
cache_id=str(cache_id),
80+
force=True,
81+
)
82+
commands = PeachFuzzer.get_fuzzed_command(
83+
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
84+
)
85+
86+
for index, cmd in TestRun.iteration(
87+
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
88+
):
89+
with TestRun.step(f"Iteration {index + 1}"):
90+
if cmd.param in forbidden_values:
91+
TestRun.LOGGER.warning(
92+
f"Iteration skipped due to the forbidden param value {cmd.param}."
93+
)
94+
continue
95+
output = run_cmd_and_validate(
96+
cmd=cmd,
97+
value_name="Device path",
98+
is_valid=cmd.param in valid_values,
99+
)
100+
if output.exit_code == 0:
101+
with TestRun.step("Detach cache"):
102+
casadm.detach_cache(cache_id)
Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
#
2+
# Copyright(c) 2025 Huawei Technologies Co., Ltd.
3+
# SPDX-License-Identifier: BSD-3-Clause
4+
#
5+
6+
import pytest
7+
8+
from api.cas import casadm
9+
from api.cas.cache_config import (
10+
CacheMode,
11+
CacheLineSize,
12+
UnalignedIo,
13+
KernelParameters,
14+
UseIoScheduler,
15+
)
16+
from api.cas.cli import attach_cache_cmd
17+
from core.test_run import TestRun
18+
from storage_devices.disk import DiskType, DiskTypeSet
19+
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
20+
from type_def.size import Unit, Size
21+
from tests.security.fuzzy.kernel.common.common import (
22+
get_fuzz_config,
23+
run_cmd_and_validate,
24+
)
25+
26+
27+
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
28+
@pytest.mark.parametrizex("cache_mode", CacheMode)
29+
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
30+
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
31+
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
32+
def test_fuzzy_attach_cache_cache_id(cache_mode, cache_line_size, unaligned_io, use_io_scheduler):
33+
"""
34+
title: Fuzzy test for casadm 'attach cache' command – cache id
35+
description: |
36+
Using Peach Fuzzer check ability of handling wrong cache id in 'attach cache' command.
37+
pass_criteria:
38+
- System did not crash
39+
"""
40+
with TestRun.step("Create partition on cache device"):
41+
cache_disk = TestRun.disks["cache"]
42+
cache_disk.create_partitions([Size(400, Unit.MebiByte)])
43+
44+
with TestRun.step("Start cache"):
45+
cache = casadm.start_cache(
46+
cache_dev=cache_disk.partitions[0],
47+
cache_mode=cache_mode,
48+
cache_line_size=cache_line_size,
49+
cache_id=1,
50+
force=True,
51+
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
52+
)
53+
54+
with TestRun.step("Detach cache"):
55+
cache.detach()
56+
57+
with TestRun.step("Prepare PeachFuzzer"):
58+
fuzz_config = get_fuzz_config("cache_id.yml")
59+
PeachFuzzer.generate_config(fuzz_config)
60+
base_cmd = attach_cache_cmd(
61+
cache_dev=cache_disk.partitions[0].path,
62+
cache_id="{param}",
63+
force=True,
64+
)
65+
commands = PeachFuzzer.get_fuzzed_command(
66+
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
67+
)
68+
69+
for index, cmd in TestRun.iteration(
70+
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
71+
):
72+
with TestRun.step(f"Iteration {index + 1}"):
73+
output = run_cmd_and_validate(
74+
cmd=cmd,
75+
value_name="Cache id",
76+
is_valid=__is_valid(cmd.param),
77+
)
78+
if output.exit_code == 0:
79+
with TestRun.step("Detach cache"):
80+
casadm.detach_cache(cache_id=int(cmd.param))
81+
82+
83+
def __is_valid(parameter):
84+
try:
85+
value = int(parameter)
86+
except ValueError:
87+
return False
88+
return value == 1
Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
#
2+
# Copyright(c) 2025 Huawei Technologies Co., Ltd.
3+
# SPDX-License-Identifier: BSD-3-Clause
4+
#
5+
6+
import pytest
7+
8+
from api.cas import casadm
9+
from api.cas.cache_config import (
10+
CacheMode,
11+
CacheLineSize,
12+
UnalignedIo,
13+
KernelParameters,
14+
UseIoScheduler,
15+
)
16+
from api.cas.cli import attach_cache_cmd
17+
from core.test_run import TestRun
18+
from storage_devices.disk import DiskType, DiskTypeSet
19+
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
20+
from test_tools.udev import Udev
21+
from type_def.size import Unit, Size
22+
from tests.security.fuzzy.kernel.common.common import (
23+
get_fuzz_config,
24+
run_cmd_and_validate,
25+
get_cmd,
26+
)
27+
28+
29+
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
30+
@pytest.mark.parametrizex("cache_mode", CacheMode)
31+
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
32+
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
33+
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
34+
def test_fuzzy_attach_cache_flags(cache_mode, cache_line_size, unaligned_io, use_io_scheduler):
35+
"""
36+
title: Fuzzy test for casadm 'attach cache' command – flags.
37+
description: |
38+
Using Peach Fuzzer check ability of handling wrong flags in 'attach cache' command.
39+
pass_criteria:
40+
- System did not crash
41+
"""
42+
43+
cache_id = 1
44+
45+
with TestRun.step("Create partition on cache device"):
46+
cache_disk = TestRun.disks["cache"]
47+
cache_disk.create_partitions([Size(400, Unit.MebiByte)])
48+
49+
with TestRun.step("Disable udev"):
50+
Udev.disable()
51+
52+
with TestRun.step("Start cache"):
53+
cache = casadm.start_cache(
54+
cache_dev=cache_disk.partitions[0],
55+
cache_mode=cache_mode,
56+
cache_line_size=cache_line_size,
57+
cache_id=cache_id,
58+
force=True,
59+
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
60+
)
61+
62+
with TestRun.step("Detach cache"):
63+
cache.detach()
64+
65+
with TestRun.step("Prepare PeachFuzzer"):
66+
valid_values = ["", "--force", "-f"]
67+
fuzz_config = get_fuzz_config("flags.yml")
68+
PeachFuzzer.generate_config(fuzz_config)
69+
parameters = PeachFuzzer.generate_peach_fuzzer_parameters(TestRun.usr.fuzzy_iter_count)
70+
71+
for index, parameter in TestRun.iteration(
72+
enumerate(parameters), f"Run command {TestRun.usr.fuzzy_iter_count} times"
73+
):
74+
with TestRun.step(f"Iteration {index + 1}"):
75+
param = parameter.decode("ascii", "ignore").rstrip()
76+
base_cmd = attach_cache_cmd(
77+
cache_dev=cache_disk.partitions[0].path,
78+
cache_id=str(cache_id),
79+
force=False,
80+
)
81+
82+
base_cmd = f"{base_cmd.strip()} {param}"
83+
84+
cmd = get_cmd(base_cmd, param.encode("ascii"))
85+
86+
output = run_cmd_and_validate(
87+
cmd=cmd,
88+
value_name="Flag",
89+
is_valid=param in valid_values,
90+
)
91+
if output.exit_code == 0:
92+
with TestRun.step("Detach cache"):
93+
casadm.detach_cache(cache_id=cache_id)
Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
#
2+
# Copyright(c) 2025 Huawei Technologies Co., Ltd.
3+
# SPDX-License-Identifier: BSD-3-Clause
4+
#
5+
6+
import pytest
7+
8+
from api.cas import casadm
9+
from api.cas.cache_config import (
10+
CacheMode,
11+
CacheLineSize,
12+
UnalignedIo,
13+
KernelParameters,
14+
UseIoScheduler,
15+
)
16+
from api.cas.cli import detach_cache_cmd
17+
from core.test_run import TestRun
18+
from storage_devices.disk import DiskType, DiskTypeSet
19+
from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
20+
from type_def.size import Unit, Size
21+
from tests.security.fuzzy.kernel.common.common import (
22+
get_fuzz_config,
23+
run_cmd_and_validate,
24+
)
25+
26+
27+
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
28+
@pytest.mark.parametrizex("cache_mode", CacheMode)
29+
@pytest.mark.parametrizex("cache_line_size", CacheLineSize)
30+
@pytest.mark.parametrizex("unaligned_io", UnalignedIo)
31+
@pytest.mark.parametrizex("use_io_scheduler", UseIoScheduler)
32+
def test_fuzzy_detach_cache_cache_id(cache_mode, cache_line_size, unaligned_io, use_io_scheduler):
33+
"""
34+
title: Fuzzy test for casadm 'detach cache' command – cache id
35+
description: |
36+
Using Peach Fuzzer check ability of handling wrong cache id in 'detach cache' command.
37+
pass_criteria:
38+
- System did not crash
39+
"""
40+
with TestRun.step("Create partition on cache device"):
41+
cache_disk = TestRun.disks["cache"]
42+
cache_disk.create_partitions([Size(400, Unit.MebiByte)])
43+
44+
with TestRun.step("Start cache"):
45+
cache = casadm.start_cache(
46+
cache_dev=cache_disk.partitions[0],
47+
cache_mode=cache_mode,
48+
cache_line_size=cache_line_size,
49+
cache_id=1,
50+
force=True,
51+
kernel_params=KernelParameters(unaligned_io, use_io_scheduler),
52+
)
53+
54+
with TestRun.step("Prepare PeachFuzzer"):
55+
fuzz_config = get_fuzz_config("cache_id.yml")
56+
PeachFuzzer.generate_config(fuzz_config)
57+
base_cmd = detach_cache_cmd(
58+
cache_id="{param}"
59+
)
60+
commands = PeachFuzzer.get_fuzzed_command(
61+
command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
62+
)
63+
64+
for index, cmd in TestRun.iteration(
65+
enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
66+
):
67+
with TestRun.step(f"Iteration {index + 1}"):
68+
output = run_cmd_and_validate(
69+
cmd=cmd,
70+
value_name="Cache id",
71+
is_valid=__is_valid(cmd.param),
72+
)
73+
if output.exit_code == 0:
74+
with TestRun.step("Attach cache"):
75+
cache.attach(device=cache_disk.partitions[0], force=True)
76+
77+
78+
def __is_valid(parameter):
79+
try:
80+
value = int(parameter)
81+
except ValueError:
82+
return False
83+
return value == 1

0 commit comments

Comments
 (0)