nvme/nvme-cli
1
0
Fork 0
Code Issues Activity

Adding upstream version 2.7.1.

Browse source 
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-16 12:25:41 +01:00
parent 04338f02fe
commit d6e1a5d456
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
663 changed files with 15529 additions and 6994 deletions
Download patch file Download diff file Expand all files Collapse all files

2
tests/meson.build
View file

@ -95,5 +95,5 @@ if autopep8.found() and isort.found()
command : [python, linter_script, 'format'],
)
else
message('autopep8 or isort not found. Python formating disabled')
message('autopep8 or isort not found. Python formatting disabled')
endif

2
tests/nvme_attach_detach_ns_test.py
View file

@ -64,7 +64,7 @@ class TestNVMeAttachDetachNSCmd(TestNVMe):
Post Section for TestNVMeAttachDetachNSCmd
- Create primary namespace.
- Atttach it to controller.
- Attach it to controller.
- Call super class's destructor.
"""
self.assertEqual(self.create_and_validate_ns(self.default_nsid,

92
tests/nvme_copy_test.py
View file

@ -4,56 +4,110 @@
#
# Copyright (c) 2022 Samsung Electronics Co., Ltd. All Rights Reserved.
#
# Author: Arunpandian J <apj.arun@samsung.com>
# Authors: Arunpandian J <apj.arun@samsung.com>
# Joy Gu <jgu@purestorage.com>
"""
NVMe Copy Testcase:-
1. Issue copy command on set of block; shall pass.
2. If cross-namespace copy formats are supported, enable and test
cross-namespace copy formats.
"""
import subprocess
from nvme_test import TestNVMe
class TestNVMeCopy(TestNVMe):
"""
Represents NVMe Verify testcase.
Represents NVMe Copy testcase.
- Attributes:
- start_block : starting block of to verify operation.
- range : Range of blocks for DSM operation.
- slbs : 64-bit addr of first block per range
- ocfs : optional copy formats supported
- host_behavior_data : host behavior support data to restore during teardown
- test_log_dir : directory for logs, temp files.
"""
def setUp(self):
""" Pre Section for TestNVMeCopy """
super().setUp()
self.start_block = 0
self.range = 1
self.slbs = 1
self.namespace = 1
print("\nSetting up test...")
self.ocfs = self.get_ocfs()
cross_namespace_copy = self.ocfs & 0xc
if cross_namespace_copy:
# get host behavior support data
get_features_cmd = ["nvme", "get-feature", self.ctrl, "--feature-id=0x16", "--data-len=512", "-b"]
print("Running command:", " ".join(get_features_cmd))
self.host_behavior_data = subprocess.check_output(get_features_cmd)
# enable cross-namespace copy formats
if self.host_behavior_data[4] & cross_namespace_copy:
# skip if already enabled
print("Cross-namespace copy already enabled, skipping set-features")
self.host_behavior_data = None
else:
data = self.host_behavior_data[:4] + cross_namespace_copy.to_bytes(2, 'little') + self.host_behavior_data[6:]
set_features_cmd = ["nvme", "set-feature", self.ctrl, "--feature-id=0x16", "--data-len=512"]
print("Running command:", " ".join(set_features_cmd))
proc = subprocess.Popen(set_features_cmd,
stdout=subprocess.PIPE,
stdin=subprocess.PIPE)
proc.communicate(input=data)
self.assertEqual(proc.returncode, 0, "Failed to enable cross-namespace copy formats")
get_ns_id_cmd = ["nvme", "get-ns-id", self.ns1]
print("Running command:", " ".join(get_ns_id_cmd))
output = subprocess.check_output(get_ns_id_cmd)
self.ns1_nsid = int(output.decode().strip().split(':')[-1])
self.setup_log_dir(self.__class__.__name__)
def tearDown(self):
""" Post Section for TestNVMeCopy """
print("Tearing down test...")
if self.host_behavior_data:
# restore saved host behavior support data
set_features_cmd = ["nvme", "set-feature", self.ctrl, "--feature-id=0x16", "--data-len=512"]
print("Running command:", " ".join(set_features_cmd))
proc = subprocess.Popen(set_features_cmd,
stdout=subprocess.PIPE,
stdin=subprocess.PIPE)
proc.communicate(input=self.host_behavior_data)
super().tearDown()
def copy(self):
def copy(self, sdlba, blocks, slbs, **kwargs):
""" Wrapper for nvme copy
- Args:
- None
- sdlba : destination logical block address
- blocks : number of logical blocks (0-based)
- slbs : source range logical block address
- descriptor_format : copy descriptor format (optional)
- snsids : source namespace id (optional)
- sopts : source options (optional)
- Returns:
- return code for nvme copy command.
- None
"""
copy_cmd = "nvme copy " + self.ctrl + \
" --namespace-id=" + str(self.namespace) + \
" --sdlba=" + str(self.start_block) + \
" --blocks=" + str(self.range) + \
" --slbs=" + str(self.range)
return self.exec_cmd(copy_cmd)
# skip if descriptor format not supported (default format is 0)
desc_format = kwargs.get("descriptor_format", 0)
if not self.ocfs & (1 << desc_format):
print(f"Skip copy because descriptor format {desc_format} is not supported")
return
# build copy command
copy_cmd = f"nvme copy {self.ns1} --format={desc_format} --sdlba={sdlba} --blocks={blocks} --slbs={slbs}"
if "snsids" in kwargs:
copy_cmd += f" --snsids={kwargs['snsids']}"
if "sopts" in kwargs:
copy_cmd += f" --sopts={kwargs['sopts']}"
# run and assert success
print("Running command:", copy_cmd)
self.assertEqual(self.exec_cmd(copy_cmd), 0)
def test_copy(self):
""" Testcase main """
self.assertEqual(self.copy(), 0)
print("Running test...")
self.copy(0, 1, 2, descriptor_format=0)
self.copy(0, 1, 2, descriptor_format=1)
self.copy(0, 1, 2, descriptor_format=2, snsids=self.ns1_nsid)
self.copy(0, 1, 2, descriptor_format=2, snsids=self.ns1_nsid, sopts=0)
self.copy(0, 1, 2, descriptor_format=3, snsids=self.ns1_nsid)
self.copy(0, 1, 2, descriptor_format=3, snsids=self.ns1_nsid, sopts=0)

2
tests/nvme_create_max_ns_test.py
View file

@ -66,7 +66,7 @@ class TestNVMeCreateMaxNS(TestNVMe):
Post Section for TestNVMeAttachDetachNSCmd
- Create primary namespace.
- Atttach it to controller.
- Attach it to controller.
- Call super class's destructor.
"""
self.assertEqual(self.create_and_validate_ns(self.default_nsid,

2
tests/nvme_format_test.py
View file

@ -81,7 +81,7 @@ class TestNVMeFormatCmd(TestNVMe):
Post Section for TestNVMeFormatCmd
- Create primary namespace.
- Atttach it to controller.
- Attach it to controller.
- Call super class's destructor.
"""
self.assertEqual(self.create_and_validate_ns(self.default_nsid,

2
tests/nvme_get_features_test.py
View file

@ -47,7 +47,7 @@ class TestNVMeGetMandatoryFeatures(TestNVMe):
- Attributes:
- feature_id_list : list of the mandatory features.
- get_vector_list_cmd : vector list collection for 09h.
- vector_list_len : numer of the interrupt vectors.
- vector_list_len : number of the interrupt vectors.
"""
def setUp(self):

26
tests/nvme_test.py
View file

@ -39,7 +39,7 @@ from nvme_test_logger import TestNVMeLogger
class TestNVMe(unittest.TestCase):
"""
Represents a testcase, each testcase shuold inherit this
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.
@ -58,11 +58,13 @@ class TestNVMe(unittest.TestCase):
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()
self.validate_pci_device()
if self.do_validate_pci_device:
self.validate_pci_device()
def tearDown(self):
""" Post Section for TestNVMe. """
@ -70,7 +72,7 @@ class TestNVMe(unittest.TestCase):
shutil.rmtree(self.log_dir, ignore_errors=True)
def validate_pci_device(self):
""" Validate underlaying device belogs to pci subsystem.
""" Validate underlying device belongs to pci subsystem.
- Args:
- None
- Returns:
@ -93,6 +95,7 @@ class TestNVMe(unittest.TestCase):
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:
@ -162,7 +165,7 @@ class TestNVMe(unittest.TestCase):
return ctrl_id
def get_ns_list(self):
""" Wrapper for extrating the namespace list.
""" Wrapper for extracting the namespace list.
- Args:
- None
- Returns:
@ -181,7 +184,7 @@ class TestNVMe(unittest.TestCase):
return ns_list
def get_max_ns(self):
""" Wrapper for extracting maximum number of namspaces supported.
""" Wrapper for extracting maximum number of namespaces supported.
- Args:
- None
- Returns:
@ -228,6 +231,19 @@ class TestNVMe(unittest.TestCase):
print(ncap)
return int(ncap)
def get_ocfs(self):
""" Wrapper for extracting optional copy formats supported
- Args:
- None
- Returns:
- Optional Copy Formats Supported
"""
pattern = re.compile(r'^ocfs\s*: 0x[0-9a-fA-F]+$')
output = subprocess.check_output(["nvme", "id-ctrl", self.ctrl], encoding='utf-8')
ocfs_line = next(line for line in output.splitlines() if pattern.match(line))
ocfs = ocfs_line.split(":")[1].strip()
return int(ocfs, 16)
def get_format(self):
""" Wrapper for extracting format.
- Args: