Tests for promotion policy

Signed-off-by: Jan Musial <jan.musial@intel.com>

tests/functional/pyocf/types/cache.py

@@ -100,6 +100,11 @@ class PromotionPolicy(IntEnum):
     DEFAULT = ALWAYS
 
 
+class NhitParams(IntEnum):
+    INSERTION_THRESHOLD = 0
+    TRIGGER_THRESHOLD = 1
+
+
 class CleaningPolicy(IntEnum):
     NOP = 0
     ALRU = 1
@@ -251,6 +256,21 @@ class Cache:
         if status:
             raise OcfError("Error setting promotion policy", status)
 
+    def get_promotion_policy_param(self, param_id):
+        self.read_lock()
+
+        param_value = c_uint64()
+
+        status = self.owner.lib.ocf_mngt_cache_promotion_get_param(
+            self.cache_handle, param_id, byref(param_value)
+        )
+
+        self.read_unlock()
+        if status:
+            raise OcfError("Error getting promotion policy parameter", status)
+
+        return param_value
+
     def set_promotion_policy_param(self, param_id, param_value):
         self.write_lock()
 
@@ -484,6 +504,7 @@ class Cache:
                 "state": cache_info.state,
                 "eviction_policy": EvictionPolicy(cache_info.eviction_policy),
                 "cleaning_policy": CleaningPolicy(cache_info.cleaning_policy),
+                "promotion_policy": PromotionPolicy(cache_info.promotion_policy),
                 "cache_line_size": line_size,
                 "flushed": CacheLines(cache_info.flushed, line_size),
                 "core_count": cache_info.core_count,

tests/functional/pyocf/types/shared.py

@@ -54,6 +54,21 @@ class OcfCompletion:
     management API.
     """
 
+    class CompletionResult:
+        def __init__(self, completion_args):
+            self.completion_args = {
+                x[0]: i for i, x in enumerate(completion_args)
+            }
+            self.results = None
+            self.arg_types = [x[1] for x in completion_args]
+
+        def __getitem__(self, key):
+            try:
+                position = self.completion_args[key]
+                return self.results[position]
+            except KeyError:
+                raise KeyError(f"No completion argument {key} specified")
+
     def __init__(self, completion_args: list):
         """
         Provide ctypes arg list, and optionally index of status argument in
@@ -63,19 +78,14 @@ class OcfCompletion:
             for OCF completion function
         """
         self.e = Event()
-        self.completion_args = completion_args
+        self.results = OcfCompletion.CompletionResult(completion_args)
         self._as_parameter_ = self.callback
-        self.results = None
 
     @property
     def callback(self):
-        arg_types = list(list(zip(*self.completion_args))[1])
-
-        @CFUNCTYPE(c_void_p, *arg_types)
+        @CFUNCTYPE(c_void_p, *self.results.arg_types)
         def complete(*args):
-            self.results = {}
-            for i, arg in enumerate(args):
-                self.results[self.completion_args[i][0]] = arg
+            self.results.results = args
             self.e.set()
 
         return complete

tests/functional/pyocf/types/stats/cache.py

@@ -29,6 +29,7 @@ class CacheInfo(Structure):
         ("state", c_uint8),
         ("eviction_policy", c_uint32),
         ("cleaning_policy", c_uint32),
+        ("promotion_policy", c_uint32),
         ("cache_line_size", c_uint64),
         ("flushed", c_uint32),
         ("core_count", c_uint32),

tests/functional/pyocf/types/stats/shared.py

@@ -7,7 +7,7 @@ from ctypes import c_uint64, c_uint32, Structure
 
 
 class _Stat(Structure):
-    _fields_ = [("value", c_uint64), ("permil", c_uint64)]
+    _fields_ = [("value", c_uint64), ("fraction", c_uint64)]
 
 
 class OcfStatsReq(Structure):

tests/functional/pyocf/utils.py

@@ -6,8 +6,15 @@
 from ctypes import string_at
 
 
-def print_buffer(buf, length, offset=0, width=16, ignore=0,
-                 stop_after_count_ignored=0, print_fcn=print):
+def print_buffer(
+    buf,
+    length,
+    offset=0,
+    width=16,
+    ignore=0,
+    stop_after_count_ignored=0,
+    print_fcn=print,
+):
     end = int(offset) + int(length)
     offset = int(offset)
     ignored_lines = 0
@@ -20,15 +27,25 @@ def print_buffer(buf, length, offset=0, width=16, ignore=0,
         byteline = ""
         asciiline = ""
         if not any(x != ignore for x in cur_line):
-            if stop_after_count_ignored and ignored_lines > stop_after_count_ignored:
-                print_fcn("<{} bytes of '0x{:02X}' encountered, stopping>".
-                          format(stop_after_count_ignored * width, ignore))
+            if (
+                stop_after_count_ignored
+                and ignored_lines > stop_after_count_ignored
+            ):
+                print_fcn(
+                    "<{} bytes of '0x{:02X}' encountered, stopping>".format(
+                        stop_after_count_ignored * width, ignore
+                    )
+                )
                 return
             ignored_lines += 1
             continue
 
         if ignored_lines:
-            print_fcn("<{} of '0x{:02X}' bytes omitted>".format(ignored_lines * width, ignore))
+            print_fcn(
+                "<{} of '0x{:02X}' bytes omitted>".format(
+                    ignored_lines * width, ignore
+                )
+            )
             ignored_lines = 0
 
         for byte in cur_line:
@@ -58,9 +75,12 @@ class Size:
 
     def __init__(self, b: int, sector_aligned: bool = False):
         if sector_aligned:
-            self.bytes = ((b + self._SECTOR_SIZE - 1) // self._SECTOR_SIZE) * self._SECTOR_SIZE
+            self.bytes = int(
+                ((b + self._SECTOR_SIZE - 1) // self._SECTOR_SIZE)
+                * self._SECTOR_SIZE
+            )
         else:
-            self.bytes = b
+            self.bytes = int(b)
 
     def __int__(self):
         return self.bytes