Merge pull request #264 from KamilLepek/new_sec

Negative tests for promotion policy

tests/functional/pyocf/types/cache.py

@@ -70,6 +70,11 @@ class CacheDeviceConfig(Structure):
     ]
 
 
+class ConfValidValues:
+    promotion_nhit_insertion_threshold_range = range(2, 1000)
+    promotion_nhit_trigger_threshold_range = range(0, 100)
+
+
 class CacheMode(IntEnum):
     WT = 0
     WB = 1
@@ -131,7 +136,6 @@ class MetadataLayout(IntEnum):
 
 
 class Cache:
-    DEFAULT_ID = 0
     DEFAULT_BACKFILL_QUEUE_SIZE = 65536
     DEFAULT_BACKFILL_UNBLOCK = 60000
     DEFAULT_PT_UNALIGNED_IO = False

tests/functional/tests/engine/test_pp.py

@@ -3,17 +3,14 @@
 # SPDX-License-Identifier: BSD-3-Clause-Clear
 #
 
-from ctypes import c_int, cast, c_void_p
-from enum import IntEnum
+from ctypes import c_int
 import pytest
 import math
 
 from pyocf.types.cache import (
     Cache,
-    CacheMode,
     PromotionPolicy,
     NhitParams,
-    CacheLineSize,
 )
 from pyocf.types.core import Core
 from pyocf.types.volume import Volume
@@ -59,7 +56,6 @@ def test_change_to_nhit_and_back_io_in_flight(pyocf_ctx):
         * no IOs should fail
     """
 
-    io_error = False
     # Step 1
     cache_device = Volume(Size.from_MiB(30))
     core_device = Volume(Size.from_MiB(30))
@@ -265,7 +261,6 @@ def test_promoted_after_hits_various_thresholds(
 
     c.wait()
 
-    stats = cache.get_stats()
     assert (
         threshold_reached_occupancy
         == cache.get_stats()["usage"]["occupancy"]["value"] - 1

tests/functional/tests/security/test_management_fuzzy.py

@@ -5,7 +5,8 @@
 
 import pytest
 
-from pyocf.types.cache import Cache, CacheMode, CleaningPolicy, AlruParams, AcpParams
+from pyocf.types.cache import Cache, CacheMode, CleaningPolicy,\
+    AlruParams, AcpParams, PromotionPolicy, NhitParams, ConfValidValues
 from pyocf.types.core import Core
 from pyocf.types.volume import Volume
 from pyocf.utils import Size as S
@@ -13,7 +14,8 @@ from tests.utils import generate_random_numbers
 from pyocf.types.shared import OcfError, CacheLineSize, SeqCutOffPolicy
 from ctypes import (
     c_uint64,
-    c_uint32
+    c_uint32,
+    c_uint8
 )
 
 
@@ -206,3 +208,105 @@ def test_neg_set_acp_param(pyocf_ctx, cm, cls):
             continue
         with pytest.raises(OcfError, match="Error setting cleaning policy param"):
             cache.set_cleaning_policy_param(CleaningPolicy.ALRU, i, 1)
+
+
+@pytest.mark.parametrize("cm", CacheMode)
+@pytest.mark.parametrize("cls", CacheLineSize)
+@pytest.mark.security
+def test_neg_set_promotion_policy(pyocf_ctx, cm, cls):
+    """
+    Test whether it is possible to set invalid param for promotion policy
+    :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
+    )
+
+    # Change to invalid promotion policy and check if failed
+    for i in generate_random_numbers(c_uint32):
+        if i in [item.value for item in PromotionPolicy]:
+            continue
+        with pytest.raises(OcfError, match="Error setting promotion policy"):
+            cache.set_promotion_policy(i)
+
+
+@pytest.mark.parametrize("cm", CacheMode)
+@pytest.mark.parametrize("cls", CacheLineSize)
+@pytest.mark.security
+def test_neg_set_nhit_promotion_policy_param(pyocf_ctx, cm, cls):
+    """
+    Test whether it is possible to set invalid promotion policy param id for nhit promotion policy
+    :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, promotion_policy=PromotionPolicy.NHIT
+    )
+
+    # Set invalid promotion policy param id and check if failed
+    for i in generate_random_numbers(c_uint8):
+        if i in [item.value for item in NhitParams]:
+            continue
+        with pytest.raises(OcfError, match="Error setting promotion policy parameter"):
+            cache.set_promotion_policy_param(i, 1)
+
+
+@pytest.mark.parametrize("cm", CacheMode)
+@pytest.mark.parametrize("cls", CacheLineSize)
+@pytest.mark.security
+def test_neg_set_nhit_promotion_policy_param_trigger(pyocf_ctx, cm, cls):
+    """
+    Test whether it is possible to set invalid promotion policy param TRIGGER_THRESHOLD for
+    nhit promotion policy
+    :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, promotion_policy=PromotionPolicy.NHIT
+    )
+
+    # Set to invalid promotion policy trigger threshold and check if failed
+    for i in generate_random_numbers(c_uint32):
+        if i in ConfValidValues.promotion_nhit_trigger_threshold_range:
+            continue
+        with pytest.raises(OcfError, match="Error setting promotion policy parameter"):
+            cache.set_promotion_policy_param(NhitParams.TRIGGER_THRESHOLD, i)
+
+
+@pytest.mark.parametrize("cm", CacheMode)
+@pytest.mark.parametrize("cls", CacheLineSize)
+@pytest.mark.security
+def test_neg_set_nhit_promotion_policy_param_threshold(pyocf_ctx, cm, cls):
+    """
+    Test whether it is possible to set invalid promotion policy param INSERTION_THRESHOLD for
+    nhit promotion policy
+    :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, promotion_policy=PromotionPolicy.NHIT
+    )
+
+    # Set to invalid promotion policy insertion threshold and check if failed
+    for i in generate_random_numbers(c_uint32):
+        if i in ConfValidValues.promotion_nhit_insertion_threshold_range:
+            continue
+        with pytest.raises(OcfError, match="Error setting promotion policy parameter"):
+            cache.set_promotion_policy_param(NhitParams.INSERTION_THRESHOLD, i)

tests/functional/tests/security/test_management_start_fuzzy.py

@@ -6,7 +6,7 @@
 import pytest
 import logging
 from tests.utils import generate_random_numbers
-from pyocf.types.cache import Cache, CacheMode, EvictionPolicy, MetadataLayout
+from pyocf.types.cache import Cache, CacheMode, EvictionPolicy, MetadataLayout, PromotionPolicy
 from pyocf.types.volume import Volume
 from pyocf.utils import Size
 from pyocf.types.shared import OcfError, CacheLineSize
@@ -154,3 +154,22 @@ def test_fuzzy_start_max_queue_size(pyocf_ctx, max_wb_queue_size, c_uint32_rando
         logger.warning(f"Test skipped for valid values: "
                        f"'max_queue_size={max_wb_queue_size}, "
                        f"queue_unblock_size={c_uint32_randomize}'.")
+
+
+@pytest.mark.security
+@pytest.mark.parametrize("cm", CacheMode)
+@pytest.mark.parametrize("cls", CacheLineSize)
+def test_fuzzy_start_promotion_policy(pyocf_ctx, c_uint32_randomize, cm, cls):
+    """
+    Test whether it is impossible to start cache with invalid promotion policy
+    :param pyocf_ctx: basic pyocf context fixture
+    :param c_uint32_randomize: promotion policy to start with
+    :param cm: cache mode value to start cache with
+    :param cls: cache line size to start cache with
+    """
+    if c_uint32_randomize not in [item.value for item in PromotionPolicy]:
+        with pytest.raises(OcfError, match="OCF_ERR_INVAL"):
+            try_start_cache(cache_mode=cm, cache_line_size=cls, promotion_policy=c_uint32_randomize)
+    else:
+        logger.warning(
+            f"Test skipped for valid promotion policy: '{c_uint32_randomize}'. ")

tests/functional/tests/utils.py

@@ -9,6 +9,7 @@ from ctypes import (
     c_uint64,
     c_uint32,
     c_uint16,
+    c_uint8,
     c_int,
     c_uint
 )
@@ -16,6 +17,7 @@ from ctypes import (
 
 def generate_random_numbers(c_type, count=1000):
     type_dict = {
+        c_uint8: [0, c_uint8(-1).value],
         c_uint16: [0, c_uint16(-1).value],
         c_uint32: [0, c_uint32(-1).value],
         c_uint64: [0, c_uint64(-1).value],