diff --git a/tests/functional/pyocf/types/core.py b/tests/functional/pyocf/types/core.py
index f1fdf968..d774ba15 100644
--- a/tests/functional/pyocf/types/core.py
+++ b/tests/functional/pyocf/types/core.py
@@ -171,8 +171,19 @@ def set_seq_cut_off_promotion(self, count):
         if status:
             raise OcfError("Error setting core seq cut off policy promotion count", status)
 
+    def flush(self):
+        self.cache.write_lock()
+
+        c = OcfCompletion([("core", c_void_p), ("priv", c_void_p), ("error", c_int)])
+        self.cache.owner.lib.ocf_mngt_core_flush(self.handle, c, None)
+        c.wait()
+        self.cache.write_unlock()
+
+        if c.results["error"]:
+            raise OcfError("Couldn't flush cache", c.results["error"])
+
     def reset_stats(self):
-        self.cache.owner.lib.ocf_core_stats_initialize(self.handle)
+        lib.ocf_core_stats_initialize(self.handle)
 
 
 lib = OcfLib.getInstance()
@@ -192,3 +203,7 @@ def reset_stats(self):
 lib.ocf_stats_collect_core.restype = c_int
 lib.ocf_core_get_info.argtypes = [c_void_p, c_void_p]
 lib.ocf_core_get_info.restype = c_int
+lib.ocf_core_stats_initialize.argtypes = [c_void_p]
+lib.ocf_core_stats_initialize.restype = c_void_p
+lib.ocf_mngt_core_flush.argtypes = [c_void_p, c_void_p, c_void_p]
+lib.ocf_mngt_core_flush.restype = c_void_p
diff --git a/tests/functional/tests/management/test_flush.py b/tests/functional/tests/management/test_flush.py
new file mode 100644
index 00000000..88b933ce
--- /dev/null
+++ b/tests/functional/tests/management/test_flush.py
@@ -0,0 +1,92 @@
+#
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import pytest
+
+from pyocf.utils import Size as S
+from pyocf.types.cache import Cache, CacheMode, CleaningPolicy
+from pyocf.types.core import Core
+from pyocf.types.volume import RamVolume, Volume
+from pyocf.types.volume_core import CoreVolume
+from pyocf.rio import Rio, ReadWrite
+
+
+def test_flush_cache(pyocf_ctx):
+    cache = Cache.start_on_device(
+        RamVolume(S.from_MiB(50)), cache_mode=CacheMode.WB, metadata_volatile=True
+    )
+    core = Core.using_device(RamVolume(S.from_MiB(100)))
+
+    cache.add_core(core)
+
+    cache.set_cleaning_policy(CleaningPolicy.NOP)
+
+    cfv = CoreVolume(core)
+    queue = core.cache.get_default_queue()
+    r = Rio().target(cfv).bs(S.from_KiB(4))
+
+    r.copy().readwrite(ReadWrite.WRITE).size(S.from_MiB(10)).run([queue])
+
+    stats = cache.get_stats()
+    assert S(stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(10)
+
+    cache.flush()
+
+    stats = cache.get_stats()
+    assert S(stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(0)
+
+
+def test_flush_core(pyocf_ctx):
+    cache = Cache.start_on_device(
+        RamVolume(S.from_MiB(50)), cache_mode=CacheMode.WB, metadata_volatile=True
+    )
+    core1 = Core.using_device(RamVolume(S.from_MiB(100)), name="core1")
+    core2 = Core.using_device(RamVolume(S.from_MiB(100)), name="core2")
+
+    cache.add_core(core1)
+    cache.add_core(core2)
+
+    cache.set_cleaning_policy(CleaningPolicy.NOP)
+
+    cfv1 = CoreVolume(core1)
+    cfv2 = CoreVolume(core2)
+    queue = cache.get_default_queue()
+
+    r1 = Rio().target(cfv1).bs(S.from_KiB(4))
+    r2 = Rio().target(cfv2).bs(S.from_KiB(4))
+
+    r1.copy().readwrite(ReadWrite.WRITE).size(S.from_MiB(10)).run([queue])
+    r2.copy().readwrite(ReadWrite.WRITE).size(S.from_MiB(10)).run([queue])
+
+    cache_stats = cache.get_stats()
+    assert S(cache_stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(20)
+
+    core1_stats = core1.get_stats()
+    assert S(core1_stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(10)
+
+    core2_stats = core2.get_stats()
+    assert S(core2_stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(10)
+
+    core1.flush()
+
+    cache_stats = cache.get_stats()
+    assert S(cache_stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(10)
+
+    core1_stats = core1.get_stats()
+    assert S(core1_stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(0)
+
+    core2_stats = core2.get_stats()
+    assert S(core2_stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(10)
+
+    core2.flush()
+
+    cache_stats = cache.get_stats()
+    assert S(cache_stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(0)
+
+    core1_stats = core1.get_stats()
+    assert S(core1_stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(0)
+
+    core2_stats = core2.get_stats()
+    assert S(core2_stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(0)