pyocf: Seq-cutoff promotion count security tests

Signed-off-by: Michal Mielewczyk <michal.mielewczyk@intel.com>

tests/functional/pyocf/types/cache.py

@@ -85,6 +85,7 @@ class ConfValidValues:
     cleaning_acp_flush_max_buffers_range = range(1, 10000)
 
     seq_cutoff_threshold_rage = range(1, 4194181)
+    seq_cutoff_promotion_range = range(1, 65535)
 
     ioclass_id_range = range(0, 32)
     ioclass_priority_range = range(-1, 255)
@@ -319,6 +320,18 @@ class Cache:
         if status:
             raise OcfError("Error setting cache seq cut off policy threshold", status)
 
+    def set_seq_cut_off_promotion(self, count: int):
+        self.write_lock()
+
+        status = self.owner.lib.ocf_mngt_core_set_seq_cutoff_promotion_count_all(
+            self.cache_handle, count
+        )
+
+        self.write_unlock()
+
+        if status:
+            raise OcfError("Error setting cache seq cut off policy promotion count", status)
+
     def get_partition_info(self, part_id: int):
         ioclass_info = IoClassInfo()
         self.read_lock()
@@ -707,6 +720,8 @@ lib.ocf_mngt_core_set_seq_cutoff_policy_all.argtypes = [c_void_p, c_uint32]
 lib.ocf_mngt_core_set_seq_cutoff_policy_all.restype = c_int
 lib.ocf_mngt_core_set_seq_cutoff_threshold_all.argtypes = [c_void_p, c_uint32]
 lib.ocf_mngt_core_set_seq_cutoff_threshold_all.restype = c_int
+lib.ocf_mngt_core_set_seq_cutoff_promotion_count_all.argtypes = [c_void_p, c_uint32]
+lib.ocf_mngt_core_set_seq_cutoff_promotion_count_all.restype = c_int
 lib.ocf_stats_collect_cache.argtypes = [
     c_void_p,
     c_void_p,

tests/functional/pyocf/types/core.py

@@ -160,11 +160,9 @@ class Core:
         status = self.cache.owner.lib.ocf_mngt_core_set_seq_cutoff_policy(
             self.handle, policy
         )
-        if status:
-            self.cache.write_unlock()
-            raise OcfError("Error setting core seq cut off policy", status)
-
         self.cache.write_unlock()
+        if status:
+            raise OcfError("Error setting core seq cut off policy", status)
 
     def set_seq_cut_off_threshold(self, threshold):
         self.cache.write_lock()
@@ -172,11 +170,19 @@ class Core:
         status = self.cache.owner.lib.ocf_mngt_core_set_seq_cutoff_threshold(
             self.handle, threshold
         )
+        self.cache.write_unlock()
         if status:
-            self.cache.write_unlock()
             raise OcfError("Error setting core seq cut off policy threshold", status)
 
+    def set_seq_cut_off_promotion(self, count):
+        self.cache.write_lock()
+
+        status = self.cache.owner.lib.ocf_mngt_core_set_seq_cutoff_promotion_count(
+            self.handle, count
+        )
         self.cache.write_unlock()
+        if status:
+            raise OcfError("Error setting core seq cut off policy promotion count", status)
 
     def reset_stats(self):
         self.cache.owner.lib.ocf_core_stats_initialize(self.handle)
@@ -229,6 +235,8 @@ lib.ocf_mngt_core_set_seq_cutoff_policy.argtypes = [c_void_p, c_uint32]
 lib.ocf_mngt_core_set_seq_cutoff_policy.restype = c_int
 lib.ocf_mngt_core_set_seq_cutoff_threshold.argtypes = [c_void_p, c_uint32]
 lib.ocf_mngt_core_set_seq_cutoff_threshold.restype = c_int
+lib.ocf_mngt_core_set_seq_cutoff_promotion_count.argtypes = [c_void_p, c_uint32]
+lib.ocf_mngt_core_set_seq_cutoff_promotion_count.restype = c_int
 lib.ocf_stats_collect_core.argtypes = [c_void_p, c_void_p, c_void_p, c_void_p, c_void_p]
 lib.ocf_stats_collect_core.restype = c_int
 lib.ocf_core_get_info.argtypes = [c_void_p, c_void_p]

tests/functional/tests/security/test_management_fuzzy.py

@@ -20,7 +20,12 @@ from pyocf.types.cache import (
 from pyocf.types.core import Core
 from pyocf.types.volume import Volume
 from pyocf.utils import Size as S
-from tests.utils.random import Range, RandomGenerator, DefaultRanges, RandomStringGenerator
+from tests.utils.random import (
+    Range,
+    RandomGenerator,
+    DefaultRanges,
+    RandomStringGenerator,
+)
 from pyocf.types.shared import OcfError, CacheLineSize, SeqCutOffPolicy
 from ctypes import c_uint64, c_uint32, c_uint8
 
@@ -132,6 +137,75 @@ def test_neg_cache_set_seq_cut_off_policy(pyocf_ctx, cm, cls):
             print(f"\n{i}")
 
 
+@pytest.mark.parametrize("cm", CacheMode)
+@pytest.mark.parametrize("cls", CacheLineSize)
+@pytest.mark.security
+def test_neg_cache_set_seq_cut_off_promotion(pyocf_ctx, cm, cls):
+    """
+    Test whether it is possible to change cache seq cut-off promotion count to invalid value
+    :param pyocf_ctx: basic pyocf context fixture
+    :param cm: cache mode we start with
+    :param cls: cache line size we start with
+    :return:
+    """
+    # Start cache device
+    cache_device = Volume(S.from_MiB(30))
+    cache = Cache.start_on_device(cache_device, cache_mode=cm, cache_line_size=cls)
+
+    # Create 2 core devices
+    core_device1 = Volume(S.from_MiB(10))
+    core1 = Core.using_device(core_device1, name="core1")
+    core_device2 = Volume(S.from_MiB(10))
+    core2 = Core.using_device(core_device2, name="core2")
+
+    # Add cores
+    cache.add_core(core1)
+    cache.add_core(core2)
+
+    # Change cache seq cut off promotion count to invalid one and check if failed
+    for i in RandomGenerator(DefaultRanges.UINT32):
+        if i in ConfValidValues.seq_cutoff_promotion_range:
+            continue
+        with pytest.raises(
+            OcfError, match="Error setting cache seq cut off policy promotion count"
+        ):
+            cache.set_seq_cut_off_promotion(i)
+            print(f"\n{i}")
+
+
+@pytest.mark.parametrize("cm", CacheMode)
+@pytest.mark.parametrize("cls", CacheLineSize)
+@pytest.mark.security
+def test_neg_core_set_seq_cut_off_promotion(pyocf_ctx, cm, cls):
+    """
+    Test whether it is possible to change core seq cut-off promotion count to invalid value
+    :param pyocf_ctx: basic pyocf context fixture
+    :param cm: cache mode we start with
+    :param cls: cache line size we start with
+    :return:
+    """
+    # Start cache device
+    cache_device = Volume(S.from_MiB(30))
+    cache = Cache.start_on_device(cache_device, cache_mode=cm, cache_line_size=cls)
+
+    # Create core device
+    core_device1 = Volume(S.from_MiB(10))
+    core1 = Core.using_device(core_device1, name="core1")
+
+    # Add core
+    cache.add_core(core1)
+
+    # Change core seq cut off promotion count to invalid one and check if failed
+    for i in RandomGenerator(DefaultRanges.UINT32):
+        if i in ConfValidValues.seq_cutoff_promotion_range:
+            continue
+        with pytest.raises(
+            OcfError, match="Error setting core seq cut off policy promotion count"
+        ):
+            core1.set_seq_cut_off_promotion(i)
+            print(f"\n{i}")
+
+
 @pytest.mark.parametrize("cm", CacheMode)
 @pytest.mark.parametrize("cls", CacheLineSize)
 @pytest.mark.security
@@ -161,7 +235,9 @@ def test_neg_cache_set_seq_cut_off_threshold(pyocf_ctx, cm, cls):
     for i in RandomGenerator(DefaultRanges.UINT32):
         if i in ConfValidValues.seq_cutoff_threshold_rage:
             continue
-        with pytest.raises(OcfError, match="Error setting cache seq cut off policy threshold"):
+        with pytest.raises(
+            OcfError, match="Error setting cache seq cut off policy threshold"
+        ):
             cache.set_seq_cut_off_threshold(i)
             print(f"\n{i}")
 
@@ -192,7 +268,9 @@ def test_neg_core_set_seq_cut_off_threshold(pyocf_ctx, cm, cls):
     for i in RandomGenerator(DefaultRanges.UINT32):
         if i in ConfValidValues.seq_cutoff_threshold_rage:
             continue
-        with pytest.raises(OcfError, match="Error setting core seq cut off policy threshold"):
+        with pytest.raises(
+            OcfError, match="Error setting core seq cut off policy threshold"
+        ):
             core.set_seq_cut_off_threshold(i)
             print(f"\n{i}")