diff --git a/t/nvmept_flush.py b/t/nvmept_flush.py
new file mode 100755
index 0000000000..590ffb4d8f
--- /dev/null
+++ b/t/nvmept_flush.py
@@ -0,0 +1,158 @@
+import os
+import sys
+import time
+import logging
+import argparse
+from pathlib import Path
+from fiotestlib import FioJobCmdTest, run_fio_tests
+
+
+class FlushTest(FioJobCmdTest):
+    def setup(self, parameters):
+        fio_args = [
+            "--name=nvmept-flush",
+            "--ioengine=io_uring_cmd",
+            "--cmd_type=nvme",
+            "--randrepeat=0",
+            f"--filename={self.fio_opts['filename']}",
+            f"--rw={self.fio_opts['rw']}",
+            f"--output={self.filenames['output']}",
+            f"--output-format={self.fio_opts['output-format']}",
+        ]
+
+        for opt in ['fixedbufs', 'nonvectored', 'force_async', 'registerfiles',
+                    'sqthread_poll', 'sqthread_poll_cpu', 'hipri', 'nowait',
+                    'time_based', 'runtime', 'verify', 'io_size', 'num_range',
+                    'iodepth', 'iodepth_batch', 'iodepth_batch_complete',
+                    'size', 'rate', 'bs', 'bssplit', 'bsrange', 'randrepeat',
+                    'buffer_pattern', 'verify_pattern', 'offset', 'fdp',
+                    'fdp_pli', 'fdp_pli_select', 'dataplacement', 'plid_select',
+                    'plids', 'dp_scheme', 'number_ios', 'read_iolog', 'fsync']:
+            if opt in self.fio_opts:
+                option = f"--{opt}={self.fio_opts[opt]}"
+                fio_args.append(option)
+
+        super().setup(fio_args)
+
+    def check_result(self):
+        super().check_result()
+
+        job = self.json_data['jobs'][0]
+
+        rw = self.fio_opts['rw']
+        fsync = self.fio_opts['fsync']
+
+        nr_write = job['write']['total_ios']
+        nr_sync = job['sync']['total_ios']
+
+        nr_sync_exp = nr_write // fsync
+
+        # The actual number of DDIR_SYNC issued might miss one DDIR_SYNC command
+        # when the last command issued was DDIR_WRITE command.
+        if not ((nr_sync == nr_sync_exp) or (nr_sync + 1 == nr_sync_exp)):
+            logging.error(f"nr_write={nr_write}, nr_sync={nr_sync}, fsync={fsync}")
+            self.passed = False
+
+TEST_LIST = [
+    {
+        "test_id": 1,
+        "fio_opts": {
+            "rw": 'read',
+            "bs": 4096,
+            "number_ios": 10,
+            "fsync": 1,
+            "output-format": "json",
+            },
+        "test_class": FlushTest,
+    },
+    {
+        "test_id": 2,
+        "fio_opts": {
+            "rw": 'write',
+            "bs": 4096,
+            "number_ios": 10,
+            "fsync": 1,
+            "output-format": "json",
+            },
+        "test_class": FlushTest,
+    },
+    {
+        "test_id": 3,
+        "fio_opts": {
+            "rw": 'readwrite',
+            "bs": 4096,
+            "number_ios": 10,
+            "fsync": 1,
+            "output-format": "json",
+            },
+        "test_class": FlushTest,
+    },
+    {
+        "test_id": 4,
+        "fio_opts": {
+            "rw": 'trimwrite',
+            "bs": 4096,
+            "number_ios": 10,
+            "fsync": 1,
+            "output-format": "json",
+            },
+        "test_class": FlushTest,
+    },
+]
+
+
+def parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('-d', '--debug', help='Enable debug messages',
+                        action='store_true')
+    parser.add_argument('-f', '--fio',
+                        help='path to file executable (e.g., ./fio)')
+    parser.add_argument('-a', '--artifact-root', help='artifact root directory')
+    parser.add_argument('-s', '--skip', nargs='+', type=int,
+                        help='list of test(s) to skip')
+    parser.add_argument('-o', '--run-only', nargs='+', type=int,
+                        help='list of test(s) to run, skipping all others')
+    parser.add_argument('--dut', help='target NVMe character device to test '
+                        '(e.g., /dev/ng0n1). '
+                        'WARNING: THIS IS A DESTRUCTIVE TEST', required=True)
+    return parser.parse_args()
+
+def main():
+    """Run test case(s) to test whether FLUSH NVMe commands are issued properly
+    with 'io_uring_cmd' ioengine.
+    """
+    args = parse_args()
+
+    if args.debug:
+        logging.basicConfig(level=logging.DEBUG)
+    else:
+        logging.basicConfig(level=logging.INFO)
+
+    if args.artifact_root:
+        artifact_root = args.artifact_root
+    else:
+        artifact_root = f"nvmept-flush-test-{time.strftime('%Y%m%d-%H%M%S')}"
+    os.mkdir(artifact_root)
+    print(f"Artifact directory is {artifact_root}")
+
+    if args.fio:
+        fio_path = str(Path(args.fio).absolute())
+    else:
+        fio_path = 'fio'
+    print(f"fio path is {fio_path}")
+
+    for test in TEST_LIST:
+        test['fio_opts']['filename'] = args.dut
+
+    test_env = {
+              'fio_path': fio_path,
+              'fio_root': str(Path(__file__).absolute().parent.parent),
+              'artifact_root': artifact_root,
+              'basename': 'nvmept-flush',
+              }
+
+    _, failed, _ = run_fio_tests(TEST_LIST, test_env, args)
+    sys.exit(failed)
+
+if __name__ == '__main__':
+    main()