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

Merge pull request #1496 from Kamoppl/kamilg/update_tests

Browse Source 
tests: update tests
This commit is contained in:
Robert Baldyga
2024-09-10 14:35:18 +02:00
committed by GitHub
parent 45c37edd39 e8bdcdae4f
commit de16763bec
23 changed files with 2129 additions and 1665 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
test/functional/tests/cli/test_cas_memory_usage.py
View File

@@ -10,69 +10,60 @@ from api.cas.cas_module import CasModule
from core.test_run import TestRun
from test_utils.size import Unit
from test_utils.os_utils import (allocate_memory,
defaultize_memory_affecting_functions,
disable_memory_affecting_functions,
drop_caches,
get_mem_free,
is_kernel_module_loaded,
load_kernel_module,
unload_kernel_module,
unmount_ramfs)
)
@pytest.mark.os_dependent
def test_insufficient_memory_for_cas_module():
"""
title: Negative test of ability to load OpenCAS kernel module with insufficient memory.
description: |
Check that OpenCAS kernel module won’t be loaded in case not enough memory is available.
pass_criteria:
- Loading OpenCAS kernel module returns error.
title: Negative test for the ability of CAS to load the kernel module with insufficient memory.
description: |
Check that the CAS kernel module won’t be loaded if enough memory is not available
pass_criteria:
- CAS module cannot be loaded with not enough memory.
- Loading CAS with not enough memory returns error.
"""
with TestRun.step("Disable caching and memory over-committing."):
with TestRun.step("Disable caching and memory over-committing"):
disable_memory_affecting_functions()
drop_caches()
with TestRun.step("Measure memory usage without OpenCAS module."):
with TestRun.step("Measure memory usage without OpenCAS module"):
if is_kernel_module_loaded(CasModule.cache.value):
unload_kernel_module(CasModule.cache.value)
available_mem_before_cas = get_mem_free()
with TestRun.step("Load OpenCAS module"):
output = load_kernel_module(CasModule.cache.value)
if output.exit_code != 0:
TestRun.fail("Cannot load OpenCAS module!")
with TestRun.step("Load CAS module"):
load_kernel_module(CasModule.cache.value)
with TestRun.step("Measure memory usage with OpenCAS module."):
with TestRun.step("Measure memory usage with CAS module"):
available_mem_with_cas = get_mem_free()
memory_used_by_cas = available_mem_before_cas - available_mem_with_cas
TestRun.LOGGER.info(
f"OpenCAS module uses {memory_used_by_cas.get_value(Unit.MiB):.2f} MiB of DRAM."
)
with TestRun.step("Unload OpenCAS module."):
with TestRun.step("Unload CAS module"):
unload_kernel_module(CasModule.cache.value)
with TestRun.step("Allocate memory leaving not enough memory for OpenCAS module."):
memory_to_leave = memory_used_by_cas * (3 / 4)
try:
allocate_memory(get_mem_free() - memory_to_leave)
except Exception as ex:
TestRun.LOGGER.error(f"{ex}")
with TestRun.step("Allocate memory, leaving not enough memory for CAS module"):
memory_to_leave = get_mem_free() - (memory_used_by_cas * (3 / 4))
allocate_memory(memory_to_leave)
TestRun.LOGGER.info(
f"Memory left for OpenCAS module: {get_mem_free().get_value(Unit.MiB):0.2f} MiB."
)
with TestRun.step(
"Try to load OpenCAS module and check if error message is printed on failure."
"Try to load OpenCAS module and check if correct error message is printed on failure"
):
output = load_kernel_module(CasModule.cache.value)
if output.stderr and output.exit_code != 0:
memory_left = get_mem_free()
TestRun.LOGGER.info(
f"Memory left for OpenCAS module: {memory_left.get_value(Unit.MiB):0.2f} MiB."
)
TestRun.LOGGER.info(f"Cannot load OpenCAS module as expected.\n{output.stderr}")
else:
TestRun.LOGGER.error("Loading OpenCAS module successfully finished, but should fail.")
with TestRun.step("Set memory options to default"):
unmount_ramfs()
defaultize_memory_affecting_functions()

57
test/functional/tests/cli/test_cleaning_policy.py
View File

@@ -1,5 +1,6 @@
#
# Copyright(c) 2022 Intel Corporation
# Copyright(c) 2024 Huawei Technologies Co., Ltd.
# SPDX-License-Identifier: BSD-3-Clause
#
@@ -16,8 +17,6 @@ from api.cas import casadm
from api.cas.cache_config import CacheMode, CleaningPolicy
from test_utils.os_utils import Udev
wait_time_s = 30
@pytest.mark.CI
@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
@@ -25,53 +24,63 @@ wait_time_s = 30
def test_cleaning_policy():
"""
Title: test_cleaning_policy
description: | The test is to see if dirty data will be removed from the Cache
after changing the cleaning policy from NOP to one that expects a flush.
description: |
The test is to see if dirty data will be removed from the Cache after changing the
cleaning policy from NOP to one that expects a flush.
pass_criteria:
- Cache is successfully populated with dirty data
- Cleaning policy is changed successfully
- There is no dirty data after the policy change
"""
with TestRun.step("Prepare devices."):
cache_disk = TestRun.disks["cache"]
cache_disk.create_partitions([Size(1, Unit.GibiByte)])
cache_dev = cache_disk.partitions[0]
wait_time = 60
core_disk = TestRun.disks["core"]
core_disk.create_partitions([Size(1, Unit.GibiByte)])
core_dev = core_disk.partitions[0]
with TestRun.step("Partition cache and core devices"):
cache_device = TestRun.disks["cache"]
core_device = TestRun.disks["core"]
cache = casadm.start_cache(cache_dev, cache_mode=CacheMode.WB)
cache.set_cleaning_policy(CleaningPolicy.nop)
core = cache.add_core(core_dev)
cache_device.create_partitions([Size(1, Unit.GibiByte)])
core_device.create_partitions([Size(2, Unit.GibiByte)])
cache_dev = cache_device.partitions[0]
core_dev = core_device.partitions[0]
with TestRun.step("Disable udev"):
Udev.disable()
with TestRun.step("Populate cache with dirty data."):
with TestRun.step(f"Start cache in Write-Back mode and set cleaning policy to NOP"):
cache = casadm.start_cache(cache_dev, cache_mode=CacheMode.WB, force=True)
cache.set_cleaning_policy(CleaningPolicy.nop)
with TestRun.step("Add core"):
core = cache.add_core(core_dev)
with TestRun.step("Populate cache with dirty data"):
fio = (
Fio()
.create_command()
.size(Size(1, Unit.MiB))
.size(cache.size)
.read_write(ReadWrite.randwrite)
.io_engine(IoEngine.libaio)
.block_size(Size(1, Unit.Blocks4096))
.direct()
.target(core.path)
)
fio.add_job("core0").target(core.path)
fio.run()
if cache.get_dirty_blocks() <= Size.zero():
TestRun.fail("Cache does not contain dirty data.")
TestRun.fail("Cache does not contain dirty data")
with TestRun.step(f"Change cleaning policy and wait up to {wait_time_s} for flush"):
with TestRun.step("Change cleaning policy"):
cache.set_cleaning_policy(CleaningPolicy.acp)
t_end = time.time() + wait_time_s
t_end = time.time() + wait_time
while time.time() < t_end:
time.sleep(1)
if cache.get_dirty_blocks() == Size.zero():
TestRun.LOGGER.info(
f"Cache flushed after {round(time.time() - (t_end - wait_time_s), 2)} seconds."
f"Cache flushed after {round(time.time() - (t_end - wait_time), 2)} seconds."
)
break
with TestRun.step("Check if dirty data exists."):
if not cache.get_dirty_blocks() == Size.zero():
TestRun.fail("There is dirty data on cache after changing cleaning policy.")
with TestRun.step("Check if cache contains dirty data"):
if cache.get_dirty_blocks() != Size.zero():
TestRun.fail("There is dirty data on cache after changing cleaning policy")