Merge pull request #634 from arutk/mopft_21.12

pyocf: management operation power failure tests

tests/functional/pyocf/types/cache.py

@@ -75,6 +75,7 @@ class CacheAttachConfig(Structure):
         ("_open_cores", c_bool),
         ("_force", c_bool),
         ("_discard_on_start", c_bool),
+        ("_volume_params", c_void_p),
     ]
 
 
@@ -348,6 +349,7 @@ class Cache:
             raise OcfError("Error retriving ioclass info", status)
 
         return {
+            "_class_id": part_id,
             "_name": ioclass_info._name.decode("ascii"),
             "_cache_mode": ioclass_info._cache_mode,
             "_priority": int(ioclass_info._priority),
@@ -402,7 +404,7 @@ class Cache:
             ioclasses_info._config[i]._name = (
                 ioclass_info._name if len(ioclass_info._name) > 0 else 0
             )
-            ioclasses_info._config[i]._prio = ioclass_info._priority
+            ioclasses_info._config[i]._priority = ioclass_info._priority
             ioclasses_info._config[i]._cache_mode = ioclass_info._cache_mode
             ioclasses_info._config[i]._max_size = ioclass_info._max_size
 
@@ -410,7 +412,7 @@ class Cache:
 
         ioclasses_info._config[part_id]._name = name.encode("utf-8")
         ioclasses_info._config[part_id]._cache_mode = int(cache_mode)
-        ioclasses_info._config[part_id]._prio = priority
+        ioclasses_info._config[part_id]._priority = priority
         ioclasses_info._config[part_id]._max_size = max_size
 
         self.write_lock()
@@ -425,7 +427,12 @@ class Cache:
             raise OcfError("Error adding partition to cache", status)
 
     def configure_device(
-        self, device, force=False, perform_test=True, cache_line_size=None
+        self,
+        device,
+        force=False,
+        perform_test=True,
+        cache_line_size=None,
+        open_cores=True,
     ):
         self.device = device
         self.device_name = device.uuid
@@ -447,15 +454,22 @@ class Cache:
             _cache_line_size=cache_line_size
             if cache_line_size
             else self.cache_line_size,
+            _open_cores=open_cores,
             _force=force,
-            _open_cores=True,
             _discard_on_start=False,
         )
 
     def attach_device(
-        self, device, force=False, perform_test=False, cache_line_size=None
+        self,
+        device,
+        force=False,
+        perform_test=False,
+        cache_line_size=None,
+        open_cores=True,
     ):
-        self.configure_device(device, force, perform_test, cache_line_size)
+        self.configure_device(
+            device, force, perform_test, cache_line_size, open_cores=open_cores
+        )
         self.write_lock()
 
         c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
@@ -483,8 +497,8 @@ class Cache:
         if c.results["error"]:
             raise OcfError("Attaching cache device failed", c.results["error"])
 
-    def load_cache(self, device):
+    def load_cache(self, device, open_cores=True):
-        self.configure_device(device)
+        self.configure_device(device, open_cores=open_cores)
         c = OcfCompletion([("cache", c_void_p), ("priv", c_void_p), ("error", c_int)])
         device.owner.lib.ocf_mngt_cache_load(
             self.cache_handle, byref(self.dev_cfg), c, None
@@ -495,12 +509,12 @@ class Cache:
             raise OcfError("Loading cache device failed", c.results["error"])
 
     @classmethod
-    def load_from_device(cls, device, name="cache"):
+    def load_from_device(cls, device, name="cache", open_cores=True):
         c = cls(name=name, owner=device.owner)
 
         c.start_cache()
         try:
-            c.load_cache(device)
+            c.load_cache(device, open_cores=open_cores)
         except:  # noqa E722
             c.stop()
             raise
@@ -686,7 +700,8 @@ class Cache:
         self.owner.lib.ocf_mngt_cache_stop(self.cache_handle, c, None)
 
         c.wait()
-        if c.results["error"]:
+        err = OcfErrorCode(-1 * c.results["error"])
+        if err != OcfErrorCode.OCF_OK and err != OcfErrorCode.OCF_ERR_WRITE_CACHE:
             self.write_unlock()
             raise OcfError("Failed stopping cache", c.results["error"])
 
@@ -698,6 +713,9 @@ class Cache:
 
         self.owner.caches.remove(self)
 
+        if err != OcfErrorCode.OCF_OK:
+            raise OcfError("Failed stopping cache", c.results["error"])
+
     def flush(self):
         self.write_lock()
 

tests/functional/pyocf/types/ctx.py

@@ -97,6 +97,7 @@ class OcfCtx:
         self.cleaner = None
         Queue._instances_ = {}
         Volume._instances_ = {}
+        Volume._uuid_ = {}
         Data._instances_ = {}
         Logger._instances_ = {}
 

tests/functional/pyocf/types/data.py

@@ -223,3 +223,6 @@ class Data:
         m = md5()
         m.update(string_at(self.handle, self.size))
         return m.hexdigest()
+
+    def get_bytes(self):
+        return string_at(self.handle, self.size)

tests/functional/pyocf/types/ioclass.py

@@ -25,7 +25,7 @@ class IoClassConfig(Structure):
         ("_max_size", c_uint32),
         ("_name", c_char_p),
         ("_cache_mode", c_int),
-        ("_prio", c_uint16),
+        ("_priority", c_uint16),
     ]
 
 

tests/functional/pyocf/types/shared.py

@@ -12,6 +12,7 @@ from ..utils import Size as S
 
 
 class OcfErrorCode(IntEnum):
+    OCF_OK = 0
     OCF_ERR_INVAL = 1000000
     OCF_ERR_AGAIN = auto()
     OCF_ERR_INTR = auto()

tests/functional/pyocf/types/volume.py

@@ -203,7 +203,7 @@ class Volume(Structure):
             return -1
 
         if volume.opened:
-            return OcfErrorCode.OCF_ERR_NOT_OPEN_EXC
+            return -OcfErrorCode.OCF_ERR_NOT_OPEN_EXC
 
         Volume._instances_[ref] = weakref.ref(volume)
 
@@ -275,7 +275,7 @@ class Volume(Structure):
 
             discard.contents._end(discard, 0)
         except:  # noqa E722
-            discard.contents._end(discard, -5)
+            discard.contents._end(discard, -OcfErrorCode.OCF_ERR_NOT_SUPP)
 
     def get_stats(self):
         return self.stats
@@ -305,40 +305,80 @@ class Volume(Structure):
 
             io.contents._end(io, 0)
         except:  # noqa E722
-            io.contents._end(io, -5)
+            io.contents._end(io, -OcfErrorCode.OCF_ERR_IO)
 
     def dump(self, offset=0, size=0, ignore=VOLUME_POISON, **kwargs):
         if size == 0:
             size = int(self.size) - int(offset)
 
-        print_buffer(
+        print_buffer(self._storage, size, ignore=ignore, **kwargs)
-            self._storage,
-            size,
-            ignore=ignore,
-            **kwargs
-        )
 
     def md5(self):
         m = md5()
         m.update(string_at(self._storage, self.size))
         return m.hexdigest()
 
+    def get_bytes(self):
+        return string_at(self._storage, self.size)
+
 
 class ErrorDevice(Volume):
-    def __init__(self, size, error_sectors: set = None, uuid=None):
+    def __init__(
+        self,
+        size,
+        error_sectors: set = None,
+        error_seq_no: dict = None,
+        armed=True,
+        uuid=None,
+    ):
         super().__init__(size, uuid)
-        self.error_sectors = error_sectors or set()
+        self.error_sectors = error_sectors
+        self.error_seq_no = error_seq_no
+        self.armed = armed
+        self.io_seq_no = {IoDir.WRITE: 0, IoDir.READ: 0}
+        self.error = False
 
     def set_mapping(self, error_sectors: set):
         self.error_sectors = error_sectors
 
     def submit_io(self, io):
-        if io.contents._addr in self.error_sectors:
+        if not self.armed:
-            io.contents._end(io, -5)
+            super().submit_io(io)
-            self.stats["errors"][io.contents._dir] += 1
+            return
+
+        direction = IoDir(io.contents._dir)
+        seq_no_match = (
+            self.error_seq_no is not None
+            and direction in self.error_seq_no
+            and self.error_seq_no[direction] <= self.io_seq_no[direction]
+        )
+        sector_match = (
+            self.error_sectors is not None and io.contents._addr in self.error_sectors
+        )
+
+        self.io_seq_no[direction] += 1
+
+        error = True
+        if self.error_seq_no is not None and not seq_no_match:
+            error = False
+        if self.error_sectors is not None and not sector_match:
+            error = False
+        if error:
+            self.error = True
+            io.contents._end(io, -OcfErrorCode.OCF_ERR_IO)
+            self.stats["errors"][direction] += 1
         else:
             super().submit_io(io)
 
+    def arm(self):
+        self.armed = True
+
+    def disarm(self):
+        self.armed = False
+
+    def error_triggered(self):
+        return self.error
+
     def reset_stats(self):
         super().reset_stats()
         self.stats["errors"] = {IoDir.WRITE: 0, IoDir.READ: 0}

tests/functional/pyocf/wrappers/ocf_core_wrappers.c

@@ -0,0 +1,14 @@
+/*
+ * Copyright(c) 2021-2022 Intel Corporation
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#include "ocf/ocf_io.h"
+#include "ocf/ocf_core.h"
+
+const struct ocf_volume_uuid *ocf_core_get_uuid_wrapper(ocf_core_t core)
+{
+	return ocf_core_get_uuid(core);
+}
+
+

tests/functional/pytest.ini

@@ -1,2 +1,5 @@
 [pytest]
+markers =
+	security: security objectives coverage
+	long: long, do not run by default
 addopts = --ignore=tests/security -m "not long"

tests/functional/tests/surprise_shutdown/test_management_surprise_shutdown.py

@@ -0,0 +1,705 @@
+# Copyright(c) 2021-2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+from ctypes import c_int, c_void_p, byref, cast, POINTER
+
+from pyocf.types.cache import (
+    Cache,
+    CacheMode,
+    CleaningPolicy,
+    SeqCutOffPolicy,
+    PromotionPolicy,
+    AlruParams,
+    AcpParams,
+    NhitParams,
+)
+from pyocf.types.data import Data
+from pyocf.types.core import Core
+from pyocf.types.volume import ErrorDevice, Volume
+from pyocf.types.io import IoDir
+from pyocf.types.ioclass import IoClassesInfo, IoClassInfo
+from pyocf.utils import Size as S
+from pyocf.types.shared import (
+    OcfCompletion,
+    CacheLineSize,
+    OcfError,
+    OcfErrorCode,
+    Uuid,
+)
+from pyocf.ocf import OcfLib
+
+mngmt_op_surprise_shutdown_test_cache_size = S.from_MiB(40)
+mngmt_op_surprise_shutdown_test_io_offset = S.from_MiB(4).B
+
+
+def ocf_write(cache, core, val, offset):
+    data = Data.from_bytes(bytes([val] * 512))
+    comp = OcfCompletion([("error", c_int)])
+    io = core.new_io(cache.get_default_queue(), offset, 512, IoDir.WRITE, 0, 0)
+    io.set_data(data)
+    io.callback = comp.callback
+    io.submit()
+    comp.wait()
+
+
+def ocf_read(cache, core, offset):
+    data = Data(byte_count=512)
+    comp = OcfCompletion([("error", c_int)])
+    io = core.new_io(cache.get_default_queue(), offset, 512, IoDir.READ, 0, 0)
+    io.set_data(data)
+    io.callback = comp.callback
+    io.submit()
+    comp.wait()
+    return data.get_bytes()[0]
+
+
+def mngmt_op_surprise_shutdown_test(
+    pyocf_ctx, mngt_func, prepare_func, consistency_check_func
+):
+    error_triggered = True
+    error_io_seq_no = 0
+
+    while error_triggered:
+        # Start cache device without error injection
+        error_io = {IoDir.WRITE: error_io_seq_no}
+        device = ErrorDevice(
+            mngmt_op_surprise_shutdown_test_cache_size, armed=False, error_seq_no=error_io
+        )
+        cache = Cache.start_on_device(device, cache_mode=CacheMode.WB)
+
+        if prepare_func:
+            prepare_func(cache)
+
+        # make sure cache state is persistent
+        cache.save()
+
+        # initiate error injection starting at write no @error_io_seq_no
+        device.arm()
+
+        # call tested management function
+        status = 0
+        try:
+            mngt_func(cache)
+        except OcfError as ex:
+            status = ex.error_code
+
+        # if error was injected we expect mngmt op error
+        error_triggered = device.error_triggered()
+        assert error_triggered == (status != 0)
+        if error_triggered:
+            assert (
+                status == OcfErrorCode.OCF_ERR_WRITE_CACHE
+                or status == OcfErrorCode.OCF_ERR_IO
+            )
+
+        # stop cache with error injection still on
+        with pytest.raises(OcfError) as ex:
+            cache.stop()
+        assert ex.value.error_code == OcfErrorCode.OCF_ERR_WRITE_CACHE
+
+        # disable error injection and load the cache
+        device.disarm()
+
+        # load cache with open_cores = False to allow consistency check to add
+        # core with WA for pyocf object management
+        cache = Cache.load_from_device(device, open_cores=False)
+
+        # run consistency check
+        if consistency_check_func is not None:
+            consistency_check_func(cache, error_triggered)
+
+        # stop the cache
+        cache.stop()
+
+        # advance error injection point
+        error_io_seq_no += 1
+
+
+# power failure during core insert
+@pytest.mark.security
+def test_surprise_shutdown_add_core(pyocf_ctx):
+    core_device = Volume(S.from_MiB(10))
+
+    def check_core(cache, error_triggered):
+        stats = cache.get_stats()
+        assert stats["conf"]["core_count"] == (0 if error_triggered else 1)
+
+    def tested_func(cache):
+        core = Core(device=core_device, try_add=False)
+        cache.add_core(core)
+
+    def check_func(cache, error_triggered):
+        check_core(cache, error_triggered)
+
+    mngmt_op_surprise_shutdown_test(pyocf_ctx, tested_func, None, check_func)
+
+
+# power failure during core removal
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_remove_core(pyocf_ctx):
+    core_device = Volume(S.from_MiB(10))
+    core = Core.using_device(core_device)
+
+    def prepare_func(cache):
+        cache.add_core(core)
+
+    def tested_func(cache):
+        cache.remove_core(core)
+
+    def check_func(cache, error_triggered):
+        stats = cache.get_stats()
+        assert stats["conf"]["core_count"] == (1 if error_triggered else 0)
+
+    mngmt_op_surprise_shutdown_test(pyocf_ctx, tested_func, prepare_func, check_func)
+
+
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_remove_core_with_data(pyocf_ctx):
+    io_offset = mngmt_op_surprise_shutdown_test_io_offset
+    core_device = Volume(S.from_MiB(10))
+    core = Core.using_device(core_device)
+
+    def prepare_func(cache):
+        cache.add_core(core)
+        ocf_write(cache, core, 0xAA, io_offset)
+
+    def tested_func(cache):
+        cache.flush()
+        cache.remove_core(core)
+
+    def check_func(cache, error_triggered):
+        stats = cache.get_stats()
+        if stats["conf"]["core_count"] == 0:
+            assert core_device.get_bytes()[io_offset] == 0xAA
+        else:
+            core = Core(device=core_device, try_add=True)
+            cache.add_core(core)
+            assert ocf_read(cache, core, io_offset) == 0xAA
+
+    mngmt_op_surprise_shutdown_test(pyocf_ctx, tested_func, prepare_func, check_func)
+
+
+# power failure during core add after previous core removed
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_swap_core(pyocf_ctx):
+    core_device_1 = Volume(S.from_MiB(10), uuid="dev1")
+    core_device_2 = Volume(S.from_MiB(10), uuid="dev2")
+    core1 = Core.using_device(core_device_1, name="core1")
+    core2 = Core.using_device(core_device_2, name="core2")
+
+    def prepare(cache):
+        cache.add_core(core1)
+        cache.save()
+        cache.remove_core(core1)
+        cache.save()
+
+    def tested_func(cache):
+        cache.add_core(core2)
+
+    def check_func(cache, error_triggered):
+        stats = cache.get_stats()
+        assert stats["conf"]["core_count"] == (0 if error_triggered else 1)
+        core1_ptr = c_void_p()
+        core2_ptr = c_void_p()
+        ret1 = OcfLib.getInstance().ocf_core_get_by_name(
+            cache, "core1".encode("utf-8"), 6, byref(core1_ptr)
+        )
+        ret2 = OcfLib.getInstance().ocf_core_get_by_name(
+            cache, "core2".encode("utf-8"), 6, byref(core2_ptr)
+        )
+        assert ret1 != 0
+        if error_triggered:
+            assert ret2 != 0
+        else:
+            assert ret2 == 0
+            uuid_ptr = cast(
+                cache.owner.lib.ocf_core_get_uuid_wrapper(core2_ptr), POINTER(Uuid)
+            )
+            uuid = str(uuid_ptr.contents._data, encoding="ascii")
+            assert uuid == "dev2"
+
+    mngmt_op_surprise_shutdown_test(pyocf_ctx, tested_func, prepare, check_func)
+
+
+# power failure during core add after previous core removed
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_swap_core_with_data(pyocf_ctx):
+    core_device_1 = Volume(S.from_MiB(10), uuid="dev1")
+    core_device_2 = Volume(S.from_MiB(10), uuid="dev2")
+    core1 = Core.using_device(core_device_1, name="core1")
+    core2 = Core.using_device(core_device_2, name="core2")
+
+    def prepare(cache):
+        cache.add_core(core1)
+        cache.save()
+        ocf_write(cache, core1, 0xAA, mngmt_op_surprise_shutdown_test_io_offset)
+        cache.remove_core(core1)
+        cache.save()
+
+    def tested_func(cache):
+        cache.add_core(core2)
+
+    def check_func(cache, error_triggered):
+        stats = cache.get_stats()
+        assert stats["conf"]["core_count"] == (0 if error_triggered else 1)
+        core1_ptr = c_void_p()
+        core2_ptr = c_void_p()
+        ret1 = OcfLib.getInstance().ocf_core_get_by_name(
+            cache, "core1".encode("utf-8"), 6, byref(core1_ptr)
+        )
+        ret2 = OcfLib.getInstance().ocf_core_get_by_name(
+            cache, "core2".encode("utf-8"), 6, byref(core2_ptr)
+        )
+        assert ret1 != 0
+        if ret2 == 0:
+            uuid_ptr = cast(
+                cache.owner.lib.ocf_core_get_uuid_wrapper(core2_ptr), POINTER(Uuid)
+            )
+            uuid = str(uuid_ptr.contents._data, encoding="ascii")
+            assert uuid == "dev2"
+            core2 = Core(device=core_device_2, try_add=True, name="core2")
+            cache.add_core(core2)
+            assert (
+                ocf_read(cache, core2, mngmt_op_surprise_shutdown_test_io_offset)
+                == Volume.VOLUME_POISON
+            )
+
+    mngmt_op_surprise_shutdown_test(pyocf_ctx, tested_func, prepare, check_func)
+
+
+# make sure there are no crashes when cache start is interrupted
+#    1. is this checksum mismatch actually expected and the proper way
+#       to avoid loading improperly initialized cache?
+#    2. uuid checksum mismatch should not allow cache to load
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_start_cache(pyocf_ctx):
+    error_triggered = True
+    error_io_seq_no = 0
+
+    while error_triggered:
+        # Start cache device without error injection
+        error_io = {IoDir.WRITE: error_io_seq_no}
+        device = ErrorDevice(
+            mngmt_op_surprise_shutdown_test_cache_size, error_seq_no=error_io, armed=True
+        )
+
+        # call tested management function
+        status = 0
+        try:
+            cache = Cache.start_on_device(device, cache_mode=CacheMode.WB)
+        except OcfError as ex:
+            status = ex.error_code
+
+        # if error was injected we expect mngmt op error
+        error_triggered = device.error_triggered()
+        assert error_triggered == (status != 0)
+
+        if not error_triggered:
+            # stop cache with error injection still on
+            with pytest.raises(OcfError) as ex:
+                cache.stop()
+            assert ex.value.error_code == OcfErrorCode.OCF_ERR_WRITE_CACHE
+            break
+
+        # disable error injection and load the cache
+        device.disarm()
+        cache = None
+
+        try:
+            cache = Cache.load_from_device(device)
+        except OcfError:
+            cache = None
+
+        if cache is not None:
+            cache.stop()
+
+        # advance error injection point
+        error_io_seq_no += 1
+
+
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_stop_cache(pyocf_ctx):
+    core_device = Volume(S.from_MiB(10))
+    error_triggered = True
+    error_io_seq_no = 0
+    io_offset = mngmt_op_surprise_shutdown_test_io_offset
+
+    while error_triggered:
+        # Start cache device without error injection
+        error_io = {IoDir.WRITE: error_io_seq_no}
+        device = ErrorDevice(
+            mngmt_op_surprise_shutdown_test_cache_size, error_seq_no=error_io, armed=False
+        )
+
+        # setup cache and insert some data
+        cache = Cache.start_on_device(device, cache_mode=CacheMode.WB)
+        core = Core(device=core_device, try_add=False)
+        cache.add_core(core)
+        ocf_write(cache, core, 0xAA, io_offset)
+
+        # start error injection
+        device.arm()
+
+        try:
+            cache.stop()
+            status = OcfErrorCode.OCF_OK
+        except OcfError as ex:
+            status = ex.error_code
+
+        # if error was injected we expect mngmt op error
+        error_triggered = device.error_triggered()
+        if error_triggered:
+            assert status == OcfErrorCode.OCF_ERR_WRITE_CACHE
+        else:
+            assert status == 0
+
+        if not error_triggered:
+            break
+
+        # disable error injection and load the cache
+        device.disarm()
+        cache = None
+
+        assert core_device.get_bytes()[io_offset] == Volume.VOLUME_POISON
+
+        cache = Cache.load_from_device(device, open_cores=False)
+        stats = cache.get_stats()
+        if stats["conf"]["core_count"] == 1:
+            assert stats["usage"]["occupancy"]["value"] == 1
+            core = Core(device=core_device, try_add=True)
+            cache.add_core(core)
+            assert ocf_read(cache, core, io_offset) == 0xAA
+
+        cache.stop()
+
+        # advance error injection point
+        error_io_seq_no += 1
+
+
+@pytest.mark.security
+def test_surprise_shutdown_cache_reinit(pyocf_ctx):
+    core_device = Volume(S.from_MiB(10))
+
+    error_io = {IoDir.WRITE: 0}
+
+    io_offset = mngmt_op_surprise_shutdown_test_io_offset
+
+    error_triggered = True
+    while error_triggered:
+        # Start cache device without error injection
+        device = ErrorDevice(
+            mngmt_op_surprise_shutdown_test_cache_size, error_seq_no=error_io, armed=False
+        )
+
+        # start WB
+        cache = Cache.start_on_device(device, cache_mode=CacheMode.WB)
+        core = Core(device=core_device, try_add=False)
+        cache.add_core(core)
+
+        # insert dirty cacheline
+        ocf_write(cache, core, 0xAA, io_offset)
+
+        cache.stop()
+
+        assert core_device.get_bytes()[io_offset] == Volume.VOLUME_POISON
+
+        # start error injection
+        device.arm()
+
+        # power failure during cache re-initialization
+        try:
+            # sets force = True by default
+            cache = Cache.start_on_device(device, cache_mode=CacheMode.WB)
+            status = OcfErrorCode.OCF_OK
+        except OcfError as ex:
+            status = ex.error_code
+            cache = None
+
+        error_triggered = device.error_triggered()
+        assert error_triggered == (status == OcfErrorCode.OCF_ERR_WRITE_CACHE)
+
+        if cache:
+            with pytest.raises(OcfError) as ex:
+                cache.stop()
+            assert ex.value.error_code == OcfErrorCode.OCF_ERR_WRITE_CACHE
+
+        device.disarm()
+
+        cache = None
+        status = OcfErrorCode.OCF_OK
+        try:
+            cache = Cache.load_from_device(device)
+        except OcfError as ex:
+            status = ex.error_code
+
+        if not cache:
+            assert status == OcfErrorCode.OCF_ERR_NO_METADATA
+        else:
+            stats = cache.get_stats()
+            if stats["conf"]["core_count"] == 0:
+                assert stats["usage"]["occupancy"]["value"] == 0
+                cache.add_core(core)
+                assert ocf_read(cache, core, io_offset) == Volume.VOLUME_POISON
+
+            cache.stop()
+
+        error_io[IoDir.WRITE] += 1
+
+
+def _test_surprise_shutdown_mngmt_generic(pyocf_ctx, func):
+    core_device = Volume(S.from_MiB(10))
+    core = Core(device=core_device, try_add=False)
+
+    def prepare(cache):
+        cache.add_core(core)
+
+    def test(cache):
+        func(cache, core)
+        cache.save()
+
+    mngmt_op_surprise_shutdown_test(pyocf_ctx, test, prepare, None)
+
+
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_change_cache_mode(pyocf_ctx):
+    _test_surprise_shutdown_mngmt_generic(
+        pyocf_ctx, lambda cache, core: cache.change_cache_mode(CacheMode.WT)
+    )
+
+
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_set_cleaning_policy(pyocf_ctx):
+    core_device = Volume(S.from_MiB(10))
+    core = Core(device=core_device, try_add=False)
+
+    for c1 in CleaningPolicy:
+        for c2 in CleaningPolicy:
+
+            def prepare(cache):
+                cache.add_core(core)
+                cache.set_cleaning_policy(c1)
+                cache.save()
+
+            def test(cache):
+                cache.set_cleaning_policy(c2)
+                cache.save()
+
+            mngmt_op_surprise_shutdown_test(pyocf_ctx, test, prepare, None)
+
+
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_set_seq_cut_off_policy(pyocf_ctx):
+    core_device = Volume(S.from_MiB(10))
+    core = Core(device=core_device, try_add=False)
+
+    for s1 in SeqCutOffPolicy:
+        for s2 in SeqCutOffPolicy:
+
+            def prepare(cache):
+                cache.add_core(core)
+                cache.set_seq_cut_off_policy(s1)
+                cache.save()
+
+            def test(cache):
+                cache.set_seq_cut_off_policy(s2)
+                cache.save()
+
+            mngmt_op_surprise_shutdown_test(pyocf_ctx, test, prepare, None)
+
+
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_set_seq_cut_off_promotion(pyocf_ctx):
+    _test_surprise_shutdown_mngmt_generic(
+        pyocf_ctx, lambda cache, core: cache.set_seq_cut_off_promotion(256)
+    )
+
+
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_set_seq_cut_off_threshold(pyocf_ctx):
+    _test_surprise_shutdown_mngmt_generic(
+        pyocf_ctx, lambda cache, core: cache.set_seq_cut_off_threshold(S.from_MiB(2).B)
+    )
+
+
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_set_cleaning_policy_param(pyocf_ctx):
+    core_device = Volume(S.from_MiB(10))
+    core = Core(device=core_device, try_add=False)
+
+    for pol in CleaningPolicy:
+        if pol == CleaningPolicy.NOP:
+            continue
+        if pol == CleaningPolicy.ALRU:
+            params = AlruParams
+        elif pol == CleaningPolicy.ACP:
+            params = AcpParams
+        else:
+            # add handler for new policy here
+            assert False
+
+        for p in params:
+
+            def prepare(cache):
+                cache.add_core(core)
+                cache.set_cleaning_policy(pol)
+                cache.save()
+
+            def test(cache):
+                val = None
+                if pol == CleaningPolicy.ACP:
+                    if p == AcpParams.WAKE_UP_TIME:
+                        val = 5000
+                    elif p == AcpParams.FLUSH_MAX_BUFFERS:
+                        val = 5000
+                    else:
+                        # add handler for new param here
+                        assert False
+                elif pol == CleaningPolicy.ALRU:
+                    if p == AlruParams.WAKE_UP_TIME:
+                        val = 2000
+                    elif p == AlruParams.STALE_BUFFER_TIME:
+                        val = 2000
+                    elif p == AlruParams.FLUSH_MAX_BUFFERS:
+                        val = 5000
+                    elif p == AlruParams.ACTIVITY_THRESHOLD:
+                        val = 500000
+                    else:
+                        # add handler for new param here
+                        assert False
+                cache.set_cleaning_policy_param(pol, p, val)
+                cache.save()
+
+            mngmt_op_surprise_shutdown_test(pyocf_ctx, test, prepare, None)
+
+
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_set_promotion_policy(pyocf_ctx):
+    core_device = Volume(S.from_MiB(10))
+    core = Core(device=core_device, try_add=False)
+
+    for pp1 in PromotionPolicy:
+        for pp2 in PromotionPolicy:
+
+            def prepare(cache):
+                cache.add_core(core)
+                cache.set_promotion_policy(pp1)
+                cache.save()
+
+            def test(cache):
+                cache.set_promotion_policy(pp2)
+                cache.save()
+
+            mngmt_op_surprise_shutdown_test(pyocf_ctx, test, prepare, None)
+
+
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_set_promotion_policy_param(pyocf_ctx):
+    core_device = Volume(S.from_MiB(10))
+    core = Core(device=core_device, try_add=False)
+
+    for pp in PromotionPolicy:
+        if pp == PromotionPolicy.ALWAYS:
+            continue
+        if pp == PromotionPolicy.NHIT:
+            params = NhitParams
+        else:
+            # add handler for new policy here
+            assert False
+
+        for p in params:
+
+            def prepare(cache):
+                cache.add_core(core)
+                cache.set_promotion_policy(pp)
+                cache.save()
+
+            def test(cache):
+                val = None
+                if pp == PromotionPolicy.NHIT:
+                    if p == NhitParams.INSERTION_THRESHOLD:
+                        val = 500
+                    elif p == NhitParams.TRIGGER_THRESHOLD:
+                        val = 50
+                    else:
+                        # add handler for new param here
+                        assert False
+                cache.set_promotion_policy_param(pp, p, val)
+                cache.save()
+
+            mngmt_op_surprise_shutdown_test(pyocf_ctx, test, prepare, None)
+
+
+@pytest.mark.security
+@pytest.mark.long
+def test_surprise_shutdown_set_io_class_config(pyocf_ctx):
+    core_device = Volume(S.from_MiB(10))
+    core = Core(device=core_device, try_add=False)
+
+    class_range = range(0, IoClassesInfo.MAX_IO_CLASSES)
+    old_ioclass = [
+        {
+            "_class_id": i,
+            "_name": f"old_{i}" if i > 0 else "unclassified",
+            "_max_size": i,
+            "_priority": i,
+            "_cache_mode": int(CacheMode.WB),
+        }
+        for i in range(IoClassesInfo.MAX_IO_CLASSES)
+    ]
+    new_ioclass = [
+        {
+            "_class_id": i,
+            "_name": f"new_{i}" if i > 0 else "unclassified",
+            "_max_size": 2 * i,
+            "_priority": 2 * i,
+            "_cache_mode": int(CacheMode.WT),
+        }
+        for i in range(IoClassesInfo.MAX_IO_CLASSES)
+    ]
+    keys = old_ioclass[0].keys()
+
+    def set_io_class_info(cache, desc):
+        ioclasses_info = IoClassesInfo()
+        for i in range(IoClassesInfo.MAX_IO_CLASSES):
+            ioclasses_info._config[i]._class_id = i
+            ioclasses_info._config[i]._name = desc[i]["_name"].encode("utf-8")
+            ioclasses_info._config[i]._priority = desc[i]["_priority"]
+            ioclasses_info._config[i]._cache_mode = desc[i]["_cache_mode"]
+            ioclasses_info._config[i]._max_size = desc[i]["_max_size"]
+        OcfLib.getInstance().ocf_mngt_cache_io_classes_configure(
+            cache, byref(ioclasses_info)
+        )
+
+    def prepare(cache):
+        cache.add_core(core)
+        set_io_class_info(cache, old_ioclass)
+        cache.save()
+
+    def test(cache):
+        set_io_class_info(cache, new_ioclass)
+        cache.save()
+
+    def check(cache, error_triggered):
+        curr_ioclass = [
+            {k: info[k] for k in keys}
+            for info in [cache.get_partition_info(i) for i in class_range]
+        ]
+        assert curr_ioclass == old_ioclass or curr_ioclass == new_ioclass
+
+    mngmt_op_surprise_shutdown_test(pyocf_ctx, test, prepare, check)