diff --git a/test/functional/api/cas/cache.py b/test/functional/api/cas/cache.py
index a876848..a31db71 100644
--- a/test/functional/api/cas/cache.py
+++ b/test/functional/api/cas/cache.py
@@ -117,6 +117,10 @@ class Cache:
         sync()
         assert self.get_dirty_blocks().get_value(Unit.Blocks4096) == 0
 
+    def purge_cache(self):
+        casadm.purge_cache(cache_id=self.cache_id)
+        sync()
+
     def stop(self, no_data_flush: bool = False):
         return casadm.stop_cache(self.cache_id, no_data_flush)
 
diff --git a/test/functional/api/cas/casadm.py b/test/functional/api/cas/casadm.py
index 35759ad..9f81579 100644
--- a/test/functional/api/cas/casadm.py
+++ b/test/functional/api/cas/casadm.py
@@ -79,6 +79,20 @@ def try_add(core_device: Device, cache_id: int):
     return Core(core_device.system_path, cache_id)
 
 
+def purge_cache(cache_id: int):
+    output = TestRun.executor.run(script_purge_cache(str(cache_id)))
+    if output.exit_code != 0:
+        raise CmdException("Purge cache failed.", output)
+    return output
+
+
+def purge_core(cache_id: int, core_id: int):
+    output = TestRun.executor.run(script_purge_core(str(cache_id), str(core_id)))
+    if output.exit_code != 0:
+        raise CmdException("Purge core failed.", output)
+    return output
+
+
 def reset_counters(cache_id: int, core_id: int = None, shortcut: bool = False):
     _core_id = None if core_id is None else str(core_id)
     output = TestRun.executor.run(
diff --git a/test/functional/api/cas/cli.py b/test/functional/api/cas/cli.py
index 3a43178..be66f8d 100644
--- a/test/functional/api/cas/cli.py
+++ b/test/functional/api/cas/cli.py
@@ -24,6 +24,14 @@ def script_try_add_cmd(cache_id: str, core_dev: str):
            f"--core-device {core_dev}"
 
 
+def script_purge_cache_cmd(cache_id: str):
+    return f"{casadm_bin} --script --purge-cache --cache-id {cache_id}"
+
+
+def script_purge_core_cmd(cache_id: str, core_id: str):
+    return f"{casadm_bin} --script --purge-core --cache-id {cache_id} --core-id {core_id}"
+
+
 def remove_core_cmd(cache_id: str, core_id: str, force: bool = False, shortcut: bool = False):
     command = f" -R -i {cache_id} -j {core_id}" if shortcut \
         else f" --remove-core --cache-id {cache_id} --core-id {core_id}"
diff --git a/test/functional/api/cas/core.py b/test/functional/api/cas/core.py
index a54e6b3..c9bc65e 100644
--- a/test/functional/api/cas/core.py
+++ b/test/functional/api/cas/core.py
@@ -111,6 +111,10 @@ class Core(Device):
         sync()
         assert self.get_dirty_blocks().get_value(Unit.Blocks4096) == 0
 
+    def purge_core(self):
+        casadm.purge_core(self.cache_id, self.core_id)
+        sync()
+
     def set_seq_cutoff_parameters(self, seq_cutoff_param: SeqCutOffParameters):
         return casadm.set_param_cutoff(self.cache_id, self.core_id,
                                        seq_cutoff_param.threshold, seq_cutoff_param.policy)