summaryrefslogtreecommitdiffstats
path: root/tests/nvme_test.py
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2022-07-26 05:11:33 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2022-07-26 05:11:33 +0000
commita751023422eadf87cd8d6484878193a4914a9d85 (patch)
tree3fca0ae6325173fc2583e7d88abe4c38e6193277 /tests/nvme_test.py
parentAdding upstream version 2.0. (diff)
downloadnvme-cli-a751023422eadf87cd8d6484878193a4914a9d85.tar.xz
nvme-cli-a751023422eadf87cd8d6484878193a4914a9d85.zip
Adding upstream version 2.1~rc0.upstream/2.1_rc0
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'tests/nvme_test.py')
-rw-r--r--tests/nvme_test.py65
1 files changed, 25 insertions, 40 deletions
diff --git a/tests/nvme_test.py b/tests/nvme_test.py
index 11c89fe..d5eca18 100644
--- a/tests/nvme_test.py
+++ b/tests/nvme_test.py
@@ -1,3 +1,5 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
# Copyright (c) 2015-2016 Western Digital Corporation or its affiliates.
#
# This program is free software; you can redistribute it and/or
@@ -29,13 +31,12 @@ import stat
import subprocess
import sys
import time
+import unittest
-from nose import tools
-from nose.tools import assert_equal
from nvme_test_logger import TestNVMeLogger
-class TestNVMe(object):
+class TestNVMe(unittest.TestCase):
"""
Represents a testcase, each testcase shuold inherit this
@@ -51,24 +52,23 @@ class TestNVMe(object):
- clear_log_dir : default log directory.
"""
- def __init__(self):
+ def setUp(self):
""" Pre Section for TestNVMe. """
# common code used in various testcases.
self.ctrl = "XXX"
self.ns1 = "XXX"
self.test_log_dir = "XXX"
self.default_nsid = 0x1
- self.config_file = 'config.json'
+ self.config_file = 'tests/config.json'
self.load_config()
self.validate_pci_device()
- def __del__(self):
+ def tearDown(self):
""" Post Section for TestNVMe. """
if self.clear_log_dir is True:
shutil.rmtree(self.log_dir, ignore_errors=True)
- @tools.nottest
def validate_pci_device(self):
""" Validate underlaying device belogs to pci subsystem.
- Args:
@@ -79,9 +79,8 @@ class TestNVMe(object):
x1, x2, dev = self.ctrl.split('/')
cmd = cmd = "find /sys/devices -name \\*" + dev + " | grep -i pci"
err = subprocess.call(cmd, shell=True)
- assert_equal(err, 0, "ERROR : Only NVMe PCI subsystem is supported")
+ self.assertEqual(err, 0, "ERROR : Only NVMe PCI subsystem is supported")
- @tools.nottest
def load_config(self):
""" Load Basic test configuration.
- Args:
@@ -102,7 +101,6 @@ class TestNVMe(object):
if not os.path.exists(self.log_dir):
os.makedirs(self.log_dir)
- @tools.nottest
def setup_log_dir(self, test_name):
""" Set up the log directory for a testcase
Args:
@@ -116,14 +114,12 @@ class TestNVMe(object):
sys.stdout = TestNVMeLogger(self.test_log_dir + "/" + "stdout.log")
sys.stderr = TestNVMeLogger(self.test_log_dir + "/" + "stderr.log")
- @tools.nottest
def exec_cmd(self, cmd):
""" Wrapper for executing a shell command and return the result. """
proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE,
encoding='utf-8')
return proc.wait()
- @tools.nottest
def nvme_reset_ctrl(self):
""" Wrapper for nvme reset command.
- Args:
@@ -136,7 +132,7 @@ class TestNVMe(object):
shell=True,
stdout=subprocess.PIPE,
encoding='utf-8')
- assert_equal(err, 0, "ERROR : nvme reset failed")
+ self.assertEqual(err, 0, "ERROR : nvme reset failed")
time.sleep(5)
rescan_cmd = "echo 1 > /sys/bus/pci/rescan"
proc = subprocess.Popen(rescan_cmd,
@@ -145,9 +141,8 @@ class TestNVMe(object):
stderr=subprocess.PIPE,
encoding='utf-8')
time.sleep(5)
- assert_equal(proc.wait(), 0, "ERROR : pci rescan failed")
+ self.assertEqual(proc.wait(), 0, "ERROR : pci rescan failed")
- @tools.nottest
def get_ctrl_id(self):
""" Wrapper for extracting the controller id.
- Args:
@@ -161,12 +156,11 @@ class TestNVMe(object):
stdout=subprocess.PIPE,
encoding='utf-8')
err = proc.wait()
- assert_equal(err, 0, "ERROR : nvme list-ctrl failed")
+ self.assertEqual(err, 0, "ERROR : nvme list-ctrl failed")
line = proc.stdout.readline()
ctrl_id = line.split(":")[1].strip()
return ctrl_id
- @tools.nottest
def get_ns_list(self):
""" Wrapper for extrating the namespace list.
- Args:
@@ -180,13 +174,12 @@ class TestNVMe(object):
shell=True,
stdout=subprocess.PIPE,
encoding='utf-8')
- assert_equal(proc.wait(), 0, "ERROR : nvme list namespace failed")
+ self.assertEqual(proc.wait(), 0, "ERROR : nvme list namespace failed")
for line in proc.stdout:
ns_list.append(line.split('x')[-1])
return ns_list
- @tools.nottest
def get_max_ns(self):
""" Wrapper for extracting maximum number of namspaces supported.
- Args:
@@ -202,7 +195,7 @@ class TestNVMe(object):
stdout=subprocess.PIPE,
encoding='utf-8')
err = proc.wait()
- assert_equal(err, 0, "ERROR : reading maximum namespace count failed")
+ self.assertEqual(err, 0, "ERROR : reading maximum namespace count failed")
for line in proc.stdout:
if pattern.match(line):
@@ -211,7 +204,6 @@ class TestNVMe(object):
print(max_ns)
return int(max_ns)
- @tools.nottest
def get_ncap(self):
""" Wrapper for extracting capacity.
- Args:
@@ -227,7 +219,7 @@ class TestNVMe(object):
stdout=subprocess.PIPE,
encoding='utf-8')
err = proc.wait()
- assert_equal(err, 0, "ERROR : reading nvm capacity failed")
+ self.assertEqual(err, 0, "ERROR : reading nvm capacity failed")
for line in proc.stdout:
if pattern.match(line):
@@ -236,7 +228,6 @@ class TestNVMe(object):
print(ncap)
return int(ncap)
- @tools.nottest
def get_format(self):
""" Wrapper for extracting format.
- Args:
@@ -252,7 +243,7 @@ class TestNVMe(object):
stdout=subprocess.PIPE,
encoding='utf-8')
err = proc.wait()
- assert_equal(err, 0, "ERROR : reading nvm capacity failed")
+ self.assertEqual(err, 0, "ERROR : reading nvm capacity failed")
for line in proc.stdout:
if "in use" in line:
@@ -260,7 +251,6 @@ class TestNVMe(object):
print(nvm_format)
return int(nvm_format)
- @tools.nottest
def delete_all_ns(self):
""" Wrapper for deleting all the namespaces.
- Args:
@@ -269,16 +259,15 @@ class TestNVMe(object):
- None
"""
delete_ns_cmd = "nvme delete-ns " + self.ctrl + " -n 0xFFFFFFFF"
- assert_equal(self.exec_cmd(delete_ns_cmd), 0)
+ self.assertEqual(self.exec_cmd(delete_ns_cmd), 0)
list_ns_cmd = "nvme list-ns " + self.ctrl + " --all | wc -l"
proc = subprocess.Popen(list_ns_cmd,
shell=True,
stdout=subprocess.PIPE,
encoding='utf-8')
output = proc.stdout.read().strip()
- assert_equal(output, '0', "ERROR : deleting all namespace failed")
+ self.assertEqual(output, '0', "ERROR : deleting all namespace failed")
- @tools.nottest
def create_ns(self, nsze, ncap, flbas, dps):
""" Wrapper for creating a namespace.
- Args:
@@ -294,7 +283,6 @@ class TestNVMe(object):
" --flbas=" + str(flbas) + " --dps=" + str(dps)
return self.exec_cmd(create_ns_cmd)
- @tools.nottest
def create_and_validate_ns(self, nsid, nsze, ncap, flbas, dps):
""" Wrapper for creating and validating a namespace.
- Args:
@@ -316,7 +304,6 @@ class TestNVMe(object):
encoding='utf-8')
return err
- @tools.nottest
def attach_ns(self, ctrl_id, ns_id):
""" Wrapper for attaching the namespace.
- Args:
@@ -341,7 +328,6 @@ class TestNVMe(object):
err = 0 if stat.S_ISBLK(os.stat(self.ns1).st_mode) else 1
return err
- @tools.nottest
def detach_ns(self, ctrl_id, nsid):
""" Wrapper for detaching the namespace.
- Args:
@@ -358,7 +344,6 @@ class TestNVMe(object):
stdout=subprocess.PIPE,
encoding='utf-8')
- @tools.nottest
def delete_and_validate_ns(self, nsid):
""" Wrapper for deleting and validating that namespace is deleted.
- Args:
@@ -372,7 +357,7 @@ class TestNVMe(object):
shell=True,
stdout=subprocess.PIPE,
encoding='utf-8')
- assert_equal(err, 0, "ERROR : delete namespace failed")
+ self.assertEqual(err, 0, "ERROR : delete namespace failed")
return err
def get_smart_log(self, nsid):
@@ -389,7 +374,7 @@ class TestNVMe(object):
stdout=subprocess.PIPE,
encoding='utf-8')
err = proc.wait()
- assert_equal(err, 0, "ERROR : nvme smart log failed")
+ self.assertEqual(err, 0, "ERROR : nvme smart log failed")
for line in proc.stdout:
if "data_units_read" in line:
@@ -428,7 +413,7 @@ class TestNVMe(object):
stdout=subprocess.PIPE,
encoding='utf-8')
err = proc.wait()
- assert_equal(err, 0, "ERROR : nvme id controller failed")
+ self.assertEqual(err, 0, "ERROR : nvme id controller failed")
return err
def get_error_log(self):
@@ -445,7 +430,7 @@ class TestNVMe(object):
stdout=subprocess.PIPE,
encoding='utf-8')
err = proc.wait()
- assert_equal(err, 0, "ERROR : nvme error log failed")
+ self.assertEqual(err, 0, "ERROR : nvme error log failed")
line = proc.stdout.readline()
err_log_entry_count = int(line.split(" ")[5].strip().split(":")[1])
entry_count = 0
@@ -470,14 +455,14 @@ class TestNVMe(object):
run_io = subprocess.Popen(io_cmd, shell=True, stdout=subprocess.PIPE,
encoding='utf-8')
run_io_result = run_io.communicate()[1]
- assert_equal(run_io_result, None)
+ self.assertEqual(run_io_result, None)
io_cmd = "dd if=/dev/zero of=" + ns_path + " bs=" + \
str(block_size) + " count=10 > /dev/null 2>&1"
print(io_cmd)
run_io = subprocess.Popen(io_cmd, shell=True, stdout=subprocess.PIPE,
encoding='utf-8')
run_io_result = run_io.communicate()[1]
- assert_equal(run_io_result, None)
+ self.assertEqual(run_io_result, None)
def supp_check_id_ctrl(self, key):
""" Wrapper for support check.
@@ -493,8 +478,8 @@ class TestNVMe(object):
stdout=subprocess.PIPE,
encoding='utf-8')
err = proc.wait()
- assert_equal(err, 0, "ERROR : nvme Identify controller Data \
- structure failed")
+ self.assertEqual(err, 0, "ERROR : nvme Identify controller Data \
+ structure failed")
for line in proc.stdout:
if key in line:
key = line.replace(",", "", 1)