nvme-cli/tests/nvme_test.py

# 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
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
# MA  02110-1301, USA.
#
#   Author: Chaitanya Kulkarni <chaitanya.kulkarni@hgst.com>
#
""" Base class for all the testcases
"""

import json
import mmap
import os
import re
import shutil
import stat
import subprocess
import sys
import time
import unittest

from nvme_test_logger import TestNVMeLogger


class TestNVMe(unittest.TestCase):

    """
    Represents a testcase, each testcase should inherit this
    class or appropriate subclass which is a child of this class.

    Common utility functions used in various testcases.

        - Attributes:
            - ctrl : NVMe Controller.
            - ns1 : default namespace.
            - default_nsid : default namespace id.
            - config_file : configuration file.
            - clear_log_dir : default log directory.
    """

    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.do_validate_pci_device = True
        self.default_nsid = 0x1
        self.config_file = 'tests/config.json'

        self.load_config()
        if self.do_validate_pci_device:
            self.validate_pci_device()

    def tearDown(self):
        """ Post Section for TestNVMe. """
        if self.clear_log_dir is True:
            shutil.rmtree(self.log_dir, ignore_errors=True)
        self.create_and_attach_default_ns()

    def create_and_attach_default_ns(self):
        """ Creates a default namespace with the full capacity of the ctrls NVM
            - Args:
                - None
            - Returns:
                - None
        """
        self.dps = 0
        self.flbas = 0
        (ds, ms) = self.get_lba_format_size()
        ncap = int(self.get_ncap() / (ds+ms))
        self.nsze = ncap
        self.ncap = ncap
        self.ctrl_id = self.get_ctrl_id()
        self.delete_all_ns()
        err = self.create_and_validate_ns(self.default_nsid,
                                          self.nsze,
                                          self.ncap,
                                          self.flbas,
                                          self.dps)
        self.assertEqual(err, 0)
        self.assertEqual(self.attach_ns(self.ctrl_id, self.default_nsid), 0)

    def validate_pci_device(self):
        """ Validate underlying device belongs to pci subsystem.
            - Args:
                - None
            - Returns:
                - None
        """
        x1, x2, dev = self.ctrl.split('/')
        cmd = cmd = "find /sys/devices -name \\*" + dev + " | grep -i pci"
        err = subprocess.call(cmd, shell=True)
        self.assertEqual(err, 0, "ERROR : Only NVMe PCI subsystem is supported")

    def load_config(self):
        """ Load Basic test configuration.
            - Args:
                - None
            - Returns:
                - None
        """
        with open(self.config_file) as data_file:
            config = json.load(data_file)
            self.ctrl = config['controller']
            self.ns1 = config['ns1']
            self.log_dir = config['log_dir']
            self.do_validate_pci_device = config.get('do_validate_pci_device', self.do_validate_pci_device)
            self.clear_log_dir = False

            if self.clear_log_dir is True:
                shutil.rmtree(self.log_dir, ignore_errors=True)

            if not os.path.exists(self.log_dir):
                os.makedirs(self.log_dir)

    def setup_log_dir(self, test_name):
        """ Set up the log directory for a testcase
            Args:
              - test_name : name of the testcase.
            Returns:
              - None
        """
        self.test_log_dir = self.log_dir + "/" + test_name
        if not os.path.exists(self.test_log_dir):
            os.makedirs(self.test_log_dir)
        sys.stdout = TestNVMeLogger(self.test_log_dir + "/" + "stdout.log")
        sys.stderr = TestNVMeLogger(self.test_log_dir + "/" + "stderr.log")

    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()

    def nvme_reset_ctrl(self):
        """ Wrapper for nvme reset command.
            - Args:
                - None:
            - Returns:
                - None
        """
        nvme_reset_cmd = "nvme reset " + self.ctrl
        err = subprocess.call(nvme_reset_cmd,
                              shell=True,
                              stdout=subprocess.PIPE,
                              encoding='utf-8')
        self.assertEqual(err, 0, "ERROR : nvme reset failed")
        time.sleep(5)
        rescan_cmd = "echo 1 > /sys/bus/pci/rescan"
        proc = subprocess.Popen(rescan_cmd,
                                shell=True,
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE,
                                encoding='utf-8')
        time.sleep(5)
        self.assertEqual(proc.wait(), 0, "ERROR : pci rescan failed")

    def get_ctrl_id(self):
        """ Wrapper for extracting the first controller id.
            - Args:
                - None
            - Returns:
                - controller id.
        """
        get_ctrl_id = f"nvme list-ctrl {self.ctrl} --output-format=json"
        proc = subprocess.Popen(get_ctrl_id,
                                shell=True,
                                stdout=subprocess.PIPE,
                                encoding='utf-8')
        err = proc.wait()
        self.assertEqual(err, 0, "ERROR : nvme list-ctrl failed")
        json_output = json.loads(proc.stdout.read())
        self.assertTrue(len(json_output['ctrl_list']) > 0,
                        "ERROR : nvme list-ctrl could not find ctrl")
        return str(json_output['ctrl_list'][0]['ctrl_id'])

    def get_ns_list(self):
        """ Wrapper for extracting the namespace list.
            - Args:
                - None
            - Returns:
                - List of the namespaces.
        """
        ns_list = []
        ns_list_cmd = "nvme list-ns " + self.ctrl
        proc = subprocess.Popen(ns_list_cmd,
                                shell=True,
                                stdout=subprocess.PIPE,
                                encoding='utf-8')
        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

    def get_max_ns(self):
        """ Wrapper for extracting maximum number of namespaces supported.
            - Args:
                - None
            - Returns:
                - maximum number of namespaces supported.
        """
        pattern = re.compile("^nn[ ]+: [0-9]", re.IGNORECASE)
        max_ns = -1
        max_ns_cmd = "nvme id-ctrl " + self.ctrl
        proc = subprocess.Popen(max_ns_cmd,
                                shell=True,
                                stdout=subprocess.PIPE,
                                encoding='utf-8')
        err = proc.wait()
        self.assertEqual(err, 0, "ERROR : reading maximum namespace count failed")

        for line in proc.stdout:
            if pattern.match(line):
                max_ns = line.split(":")[1].strip()
                break
        print(max_ns)
        return int(max_ns)

    def get_lba_status_supported(self):
        """ Check if 'Get LBA Status' command is supported by the device
            - Args:
                - None
            - Returns:
                - True if 'Get LBA Status' command is supported, otherwise False
        """
        return int(self.get_id_ctrl_field_value("oacs"), 16) & (1 << 9)

    def get_lba_format_size(self):
        """ Wrapper for extracting lba format size of the given flbas
            - Args:
                - None
            - Returns:
                - lba format size as a tuple of (data_size, metadata_size) in bytes.
        """
        nvme_id_ns_cmd = f"nvme id-ns {self.ns1} --output-format=json"
        proc = subprocess.Popen(nvme_id_ns_cmd,
                                shell=True,
                                stdout=subprocess.PIPE,
                                encoding='utf-8')
        err = proc.wait()
        self.assertEqual(err, 0, "ERROR : reading id-ns")
        json_output = json.loads(proc.stdout.read())
        self.assertTrue(len(json_output['lbafs']) > self.flbas,
                        "Error : could not match the given flbas to an existing lbaf")
        lbaf_json = json_output['lbafs'][int(self.flbas)]
        ms_expo = int(lbaf_json['ms'])
        ds_expo = int(lbaf_json['ds'])
        ds = 0
        ms = 0
        if ds_expo > 0:
            ds = (1 << ds_expo)
        if ms_expo > 0:
            ms = (1 << ms_expo)
        return (ds, ms)

    def get_ncap(self):
        """ Wrapper for extracting capacity.
            - Args:
                - None
            - Returns:
                - maximum number of namespaces supported.
        """
        pattern = re.compile("^tnvmcap[ ]+: [0-9]", re.IGNORECASE)
        ncap = -1
        ncap_cmd = "nvme id-ctrl " + self.ctrl
        proc = subprocess.Popen(ncap_cmd,
                                shell=True,
                                stdout=subprocess.PIPE,
                                encoding='utf-8')
        err = proc.wait()
        self.assertEqual(err, 0, "ERROR : reading nvm capacity failed")

        for line in proc.stdout:
            if pattern.match(line):
                ncap = line.split(":")[1].strip()
                break
        print(ncap)
        return int(ncap)

    def get_id_ctrl_field_value(self, field):
        """ Wrapper for extracting id-ctrl field values
            - Args:
                - None
            - Returns:
                - Filed value of the given field
        """
        id_ctrl_cmd = f"nvme id-ctrl {self.ctrl} --output-format=json"
        proc = subprocess.Popen(id_ctrl_cmd,
                                shell=True,
                                stdout=subprocess.PIPE,
                                encoding='utf-8')
        err = proc.wait()
        self.assertEqual(err, 0, "ERROR : reading id-ctrl failed")
        json_output = json.loads(proc.stdout.read())
        self.assertTrue(field in json_output,
                        f"ERROR : reading field '{field}' failed")
        return str(json_output[field])

    def get_ocfs(self):
        """ Wrapper for extracting optional copy formats supported
            - Args:
                - None
            - Returns:
                - Optional Copy Formats Supported
        """
        return int(self.get_id_ctrl_field_value("ocfs"), 16)

    def get_format(self):
        """ Wrapper for extracting format.
            - Args:
                - None
            - Returns:
                - maximum format of namespace.
        """
        # defaulting to 4K
        nvm_format = 4096
        nvm_format_cmd = "nvme id-ns " + self.ctrl + " -n1"
        proc = subprocess.Popen(nvm_format_cmd,
                                shell=True,
                                stdout=subprocess.PIPE,
                                encoding='utf-8')
        err = proc.wait()
        self.assertEqual(err, 0, "ERROR : reading nvm capacity failed")

        for line in proc.stdout:
            if "in use" in line:
                nvm_format = 2 ** int(line.split(":")[3].split()[0])
        print(nvm_format)
        return int(nvm_format)

    def delete_all_ns(self):
        """ Wrapper for deleting all the namespaces.
            - Args:
                - None
            - Returns:
                - None
        """
        delete_ns_cmd = "nvme delete-ns " + self.ctrl + " -n 0xFFFFFFFF"
        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()
        self.assertEqual(output, '0', "ERROR : deleting all namespace failed")

    def create_ns(self, nsze, ncap, flbas, dps):
        """ Wrapper for creating a namespace.
            - Args:
                - nsze : new namespace size.
                - ncap : new namespace capacity.
                - flbas : new namespace format.
                - dps : new namespace data protection information.
            - Returns:
                - return code of the nvme create namespace command.
        """
        create_ns_cmd = "nvme create-ns " + self.ctrl + " --nsze=" + \
                        str(nsze) + " --ncap=" + str(ncap) + \
                        " --flbas=" + str(flbas) + " --dps=" + str(dps)
        return self.exec_cmd(create_ns_cmd)

    def create_and_validate_ns(self, nsid, nsze, ncap, flbas, dps):
        """ Wrapper for creating and validating a namespace.
            - Args:
                - nsid : new namespace id.
                - nsze : new namespace size.
                - ncap : new namespace capacity.
                - flbas : new namespace format.
                - dps : new namespace data protection information.
            - Returns:
                - return 0 on success, error code on failure.
        """
        err = self.create_ns(nsze, ncap, flbas, dps)
        if err == 0:
            time.sleep(2)
            id_ns_cmd = "nvme id-ns " + self.ctrl + " -n " + str(nsid)
            err = subprocess.call(id_ns_cmd,
                                  shell=True,
                                  stdout=subprocess.PIPE,
                                  encoding='utf-8')
        return err

    def attach_ns(self, ctrl_id, ns_id):
        """ Wrapper for attaching the namespace.
            - Args:
                - ctrl_id : controller id to which namespace to be attached.
                - nsid : new namespace id.
            - Returns:
                - 0 on success, error code on failure.
        """
        attach_ns_cmd = "nvme attach-ns " + self.ctrl + \
                        " --namespace-id=" + str(ns_id) + \
                        " --controllers=" + ctrl_id
        err = subprocess.call(attach_ns_cmd,
                              shell=True,
                              stdout=subprocess.PIPE,
                              encoding='utf-8')
        time.sleep(5)
        if err == 0:
            # enumerate new namespace block device
            self.nvme_reset_ctrl()
            time.sleep(5)
            # check if new namespace block device exists
            err = 0 if stat.S_ISBLK(os.stat(self.ns1).st_mode) else 1
        return err

    def detach_ns(self, ctrl_id, nsid):
        """ Wrapper for detaching the namespace.
            - Args:
                - ctrl_id : controller id to which namespace to be attached.
                - nsid : new namespace id.
            - Returns:
                - 0 on success, error code on failure.
        """
        detach_ns_cmd = "nvme detach-ns " + self.ctrl + \
                        " --namespace-id=" + str(nsid) + \
                        " --controllers=" + ctrl_id
        return subprocess.call(detach_ns_cmd,
                               shell=True,
                               stdout=subprocess.PIPE,
                               encoding='utf-8')

    def delete_and_validate_ns(self, nsid):
        """ Wrapper for deleting and validating that namespace is deleted.
            - Args:
                - nsid : new namespace id.
            - Returns:
                - 0 on success, 1 on failure.
        """
        # delete the namespace
        delete_ns_cmd = "nvme delete-ns " + self.ctrl + " -n " + str(nsid)
        err = subprocess.call(delete_ns_cmd,
                              shell=True,
                              stdout=subprocess.PIPE,
                              encoding='utf-8')
        self.assertEqual(err, 0, "ERROR : delete namespace failed")
        return err

    def get_smart_log(self, nsid):
        """ Wrapper for nvme smart-log command.
            - Args:
                - nsid : namespace id to get smart log from.
            - Returns:
                - 0 on success, error code on failure.
        """
        smart_log_cmd = "nvme smart-log " + self.ctrl + " -n " + str(nsid)
        print(smart_log_cmd)
        proc = subprocess.Popen(smart_log_cmd,
                                shell=True,
                                stdout=subprocess.PIPE,
                                encoding='utf-8')
        err = proc.wait()
        self.assertEqual(err, 0, "ERROR : nvme smart log failed")
        smart_log_output = proc.communicate()[0]
        print(f"{smart_log_output}")
        return err

    def get_id_ctrl(self, vendor=False):
        """ Wrapper for nvme id-ctrl command.
            - Args:
              - None
            - Returns:
              - 0 on success, error code on failure.
        """
        if not vendor:
            id_ctrl_cmd = "nvme id-ctrl " + self.ctrl
        else:
            id_ctrl_cmd = "nvme id-ctrl --vendor-specific " + self.ctrl
        print(id_ctrl_cmd)
        proc = subprocess.Popen(id_ctrl_cmd,
                                shell=True,
                                stdout=subprocess.PIPE,
                                encoding='utf-8')
        err = proc.wait()
        self.assertEqual(err, 0, "ERROR : nvme id controller failed")
        return err

    def get_error_log(self):
        """ Wrapper for nvme error-log command.
            - Args:
                - None
            - Returns:
                - 0 on success, error code on failure.
        """
        pattern = re.compile(r"^ Entry\[[ ]*[0-9]+\]")
        error_log_cmd = "nvme error-log " + self.ctrl
        proc = subprocess.Popen(error_log_cmd,
                                shell=True,
                                stdout=subprocess.PIPE,
                                encoding='utf-8')
        err = proc.wait()
        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
        for line in proc.stdout:
            if pattern.match(line):
                entry_count += 1

        return 0 if err_log_entry_count == entry_count else 1

    def run_ns_io(self, nsid, lbads, count=10):
        """ Wrapper to run ios on namespace under test.
            - Args:
                - lbads : LBA Data size supported in power of 2 format.
            - Returns:
                - None
        """
        block_size = mmap.PAGESIZE if int(lbads) < 9 else 2 ** int(lbads)
        ns_path = self.ctrl + "n" + str(nsid)
        io_cmd = "dd if=" + ns_path + " of=/dev/null" + " bs=" + \
                 str(block_size) + " count=" + str(count) + " > /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]
        self.assertEqual(run_io_result, None)
        io_cmd = "dd if=/dev/zero of=" + ns_path + " bs=" + \
                 str(block_size) + " count=" + str(count) + " > /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]
        self.assertEqual(run_io_result, None)

    def supp_check_id_ctrl(self, key):
        """ Wrapper for support check.
            - Args:
                - key : search key.
            - Returns:
                - value for key requested.
        """
        id_ctrl = "nvme id-ctrl " + self.ctrl
        print("\n" + id_ctrl)
        proc = subprocess.Popen(id_ctrl,
                                shell=True,
                                stdout=subprocess.PIPE,
                                encoding='utf-8')
        err = proc.wait()
        self.assertEqual(err, 0, "ERROR : nvme Identify controller Data \
                         structure failed")
        for line in proc.stdout:
            if key in line:
                key = line.replace(",", "", 1)
        print(key)
        val = (key.split(':'))[1].strip()
        return int(val, 16)