Added utilities definitions for nvme sed functionality

Includes following
1. Initilaize SED on nvme disk.
2. Revert SED on nvme disk.
3. SED lock on nvme disk with and without key.
4. SED unlock on nvme disk with and without key.
5. Changes SED password.
6. Added pexpect definition to handle multiple inputs to same command.
7. Added dd definition to disk.py which return bool based of disk read.
8. Added new values to spell check file.

Applied pep8 style cding to file

Signed-off-by: Maram Srimannarayana Murthy <msmurthy@linux.vnet.ibm.com>

Author: maramsmurthy

avocado/utils/disk.py

@@ -108,6 +108,19 @@ def delete_loop_device(device):
     return True
 
 
+def dd_read_records_device(disk, read_records="1", out_file="/tmp/data"):
+    """
+    Reads mentioned number of blocks from device to destination
+
+    :param disk: disk absolute path
+    :param read_records: Numbers of blocks to read from disk
+    :param out_file: Destination file
+    :rtype: boolean
+    """
+    cmd = f"dd count={read_records} if={disk} of={out_file}"
+    return not process.system(cmd, ignore_status=True, sudo=True)
+
+
 def get_disks():
     """
     Returns the physical "hard drives" available on this system
@@ -130,11 +143,11 @@ def get_disks():
     except json.JSONDecodeError as je:
         raise DiskError(f"Error occurred while parsing JSON data: {je}") from je
     disks = []
-    to_process = json_data.get("blockdevices", [])
-    for device in to_process:
-        disks.append(device.get("name"))
+    for device in json_data["blockdevices"]:
+        disks.append(device["name"])
         if "children" in device:
-            to_process.extend(device["children"])
+            for child in device["children"]:
+                disks.append(child["name"])
     return disks
 
 
@@ -460,8 +473,9 @@ def get_io_scheduler_list(device_name):
     :param device_name: Device  name example like sda
     :return: list of IO scheduler
     """
-    with open(__sched_path(device_name), "r", encoding="utf-8") as fl:
-        return fl.read().translate(str.maketrans("[]", " ")).split()
+    with open(__sched_path(device_name), "r", encoding="utf-8") as f:
+        names = f.read()
+    return names.translate(str.maketrans("[]", " ")).split()
 
 
 def get_io_scheduler(device_name):

avocado/utils/nvme.py

@@ -15,6 +15,7 @@
 #
 # Copyright: 2022 IBM
 # Authors : Naresh Bannoth <nbannoth@linux.vnet.ibm.com>
+#           Maram Srimannarayana Murthy<msmurthy@linux.ibm.com>
 
 
 """
@@ -26,8 +27,11 @@
 import logging
 import os
 import re
+import sys
 import time
 
+import pexpect  # pylint: disable=E0401
+
 from avocado.utils import pci, process
 
 LOGGER = logging.getLogger(__name__)
@@ -125,7 +129,8 @@ def get_current_ns_list(controller_name, shared_ns=False):
     namespaces_ids = get_current_ns_ids(controller_name)
     if shared_ns:
         subsys = get_subsystem_using_ctrl_name(controller_name)
-        controller_name = f"nvme{subsys[len('nvme-subsy'):]}"
+        controller_name = f"nvme{subsys[len('nvme-subsys'):]}"
+
     for ns_id in namespaces_ids:
         namespace_list.append(f"/dev/{controller_name}n{ns_id}")
     return namespace_list
@@ -529,3 +534,206 @@ def get_subsystem_using_ctrl_name(ctrl):
         if ctrl in ctrls:
             return get_subsys_name_with_nqn(device_nqn)
     return ""
+
+
+def get_nvme_sed_discover_parameters(namespace):
+    """
+    Fetches values from nvme SED discover command
+
+    :param namespace: NVMe namespace path
+    :rtype: dictionary
+    :raises: NvmeException on command failure
+    :rtype: dictionary
+    """
+    cmd = f"nvme sed discover {namespace}"
+    data = process.run(cmd, ignore_status=True, sudo=True, shell=True).stdout_text
+    pattern = r"\tLocking Supported:\s*(.*)\n\tLocking Feature Enabled:\s*(.*)\n\tLocked:\s*(.*)"
+    match = re.search(pattern, data, re.MULTILINE)
+    if match:
+        locking_features = {
+            "Locking Supported": match.group(1).strip(),
+            "Locking Feature Enabled": match.group(2).strip(),
+            "Locked": match.group(3).strip(),
+        }
+        return locking_features
+    return {}
+
+
+def is_lockdown_supported(namespace):
+    """
+    Fetches information based on namespace
+    Checks if SED locking is supported for the given namespace
+
+    :param namespace: NVMe namespace path
+    :rtype: boolean
+    """
+    lockdown_attr = get_nvme_sed_discover_parameters(namespace)
+    return lockdown_attr.get("Locking Supported") == "Yes"
+
+
+def is_lockdown_enabled(namespace):
+    """
+    Fetches information based on namespace
+    Checks if SED locking feature is enabled for the given namespace
+
+    :param namespace: NVMe namespace path
+    :rtype: boolean
+    """
+    lockdown_attr = get_nvme_sed_discover_parameters(namespace)
+    return lockdown_attr.get("Locking Feature Enabled") == "Yes"
+
+
+def is_drive_locked(namespace):
+    """
+    Fetches information based on namespace
+    Checks if the drive is currently locked for the given namespace
+
+    :param namespace: NVMe namespace path
+    :rtype: boolean
+    """
+    lockdown_attr = get_nvme_sed_discover_parameters(namespace)
+    return lockdown_attr.get("Locked") == "Yes"
+
+
+def initialize_sed_locking(namespace, password):
+    """
+    Enables and initializes SED feature on nvme disk
+
+    :param namespace: NVMe namespace path
+    :param password: SED password
+    """
+    if not is_lockdown_supported(namespace):
+        raise NvmeException(f"SED initialize not supported on {namespace}")
+    if is_lockdown_enabled(namespace):
+        raise NvmeException(
+            f"nvme drive {namespace} locking is enabled, can't initialize it"
+        )
+    pexpect_cmd_execution(
+        f"nvme sed initialize {namespace}",
+        [("New Password:", password), ("Re-enter New Password:", password)],
+    )
+    if not is_lockdown_enabled(namespace):
+        raise NvmeException(f"Failed to initialize nvme disk {namespace}")
+
+
+def revert_sed_locking(namespace, password, destructive=False):
+    """
+    Reverts SED locking state to factory defaults
+
+    :param namespace: NVMe namespace path
+    :param password: Current SED password
+    :raises: NvmeException if revert is not supported, drive is not initialized,
+             drive is locked, or revert operation fails
+    """
+    if not is_lockdown_supported(namespace):
+        raise NvmeException(f"Revert not supported on {namespace}")
+    if not is_lockdown_enabled(namespace):
+        raise NvmeException(
+            f"nvme drive {namespace} locking is not enabled, can't revert it"
+        )
+    if is_drive_locked(namespace):
+        raise NvmeException(f"Reverting not valid when drive is locked {namespace}")
+    if destructive:
+        pexpect_cmd_execution(
+            f"nvme sed revert -e {namespace}",
+            [
+                ("Destructive revert erases drive data. Continue (y/n)?", "y"),
+                ("Are you sure (y/n)?", "y"),
+                ("Password:", password),
+            ],
+        )
+    else:
+        pexpect_cmd_execution(f"nvme sed revert {namespace}", [("Password:", password)])
+    if is_lockdown_enabled(namespace):
+        raise NvmeException(f"Failed to revert {namespace}")
+
+
+def unlock_drive(namespace, with_pass_key=""):
+    """
+    Unlocks SED locked driver
+
+    :param namespace: NVMe namespace path
+    :param with_pass_key: Password for unlocking (if empty, no password prompt)
+    """
+    if not is_drive_locked(namespace):
+        raise NvmeException(f"Drive is not locked, unlock failed for {namespace}")
+    cmd = f"nvme sed unlock {namespace}"
+    if with_pass_key:
+        cmd = f"{cmd} -k"
+        pexpect_cmd_execution(cmd, [("Password:", with_pass_key)])
+    elif process.system(cmd, shell=True, ignore_status=True):
+        raise NvmeException(f"namespace {namespace} unlock failed")
+    if is_drive_locked(namespace):
+        raise NvmeException(f"Unlock failed for {namespace}")
+
+
+def lock_drive(namespace, with_pass_key=""):
+    """
+    SED lock enables on nvme drive
+
+    :param namespace: NVMe namespace path
+    :param with_pass_key: Password for locking (if empty, no password prompt)
+    """
+    if is_drive_locked(namespace):
+        raise NvmeException(f"namespace {namespace} already in locked state")
+    cmd = f"nvme sed lock {namespace}"
+    if with_pass_key:
+        cmd = f"{cmd} -k"
+        pexpect_cmd_execution(cmd, [("Password:", with_pass_key)])
+    elif process.system(cmd, shell=True, ignore_status=True):
+        raise NvmeException(f"namespace {namespace} lock failed")
+    if not is_drive_locked(namespace):
+        raise NvmeException(f"locking failed for {namespace}")
+
+
+def change_sed_password(namespace, pwd1, pwd2):
+    """
+    Changes the SED password for the specified namespace
+
+    :param namespace: NVMe namespace path
+    :param pwd1: Current SED password
+    :param pwd2: New SED password
+    :raises: NvmeException if password change is not supported or drive is not initialized
+    """
+    if not is_lockdown_supported(namespace):
+        raise NvmeException(f"Change password not supported on {namespace}")
+    if not is_lockdown_enabled(namespace):
+        raise NvmeException(
+            f"nvme drive {namespace} is not initialized, can't change password"
+        )
+    pexpect_cmd_execution(
+        f"nvme sed password {namespace}",
+        [
+            ("Password:", pwd1),
+            ("New Password:", pwd2),
+            ("Re-enter New Password:", pwd2),
+        ],
+    )
+
+
+def pexpect_cmd_execution(cmd, list_of_expect_sendline):
+    """
+    Execute command using pexpect with multiple expect/sendline interactions
+
+    :param cmd: Command to execute
+    :param list_of_expect_sendline: List of (expect_pattern, sendline_value) tuples
+    :raises: NvmeException on command failures
+    """
+    try:
+        LOGGER.info("Executing command using pexpect: %s", cmd)
+        pexpect_handle = pexpect.spawn(cmd)
+        pexpect_handle.log_read = sys.stdout
+        for expect, value in list_of_expect_sendline:
+            pexpect_handle.expect(expect, timeout=30)
+            pexpect_handle.sendline(value)
+            LOGGER.debug("Matched String: %s", pexpect_handle.after.strip())
+            LOGGER.debug("Pexpect output: %s", pexpect_handle.before.strip())
+            time.sleep(3)
+        pexpect_handle.close()
+        LOGGER.info("%s command executed successfully", cmd)
+    except pexpect.exceptions.TIMEOUT as e:
+        LOGGER.error("Command timed out: %s", cmd)
+        raise NvmeException(f"Command timeout: {cmd}") from e
+    except pexpect.exceptions.EOF as e:
+        LOGGER.error("Command ended unexpectedly: %s", cmd)
+        raise NvmeException(f"Command failed unexpectedly: {cmd}") from e

spell.ignore

@@ -1,3 +1,7 @@
+msmurthy
+sendline
+SED
+pexpect
 nvme
 nbannoth
 Naresh