nvme-cli/tests/nvme_copy_test.py at 7d12fdf5eed31e867747a976547596542f955627 · linux-nvme/nvme-cli · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
# SPDX-License-Identifier: GPL-2.0-or-later
#
# This file is part of nvme-cli
#
# Copyright (c) 2022 Samsung Electronics Co., Ltd. All Rights Reserved.
#
# Authors: Arunpandian J <[email protected]>
#          Joy Gu <[email protected]>

"""
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 base64

from nvme_test import TestNVMe


class TestNVMeCopy(TestNVMe):

    """
    Represents NVMe Copy testcase.
        - Attributes:
              - 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.ocfs = self.get_ocfs()
        self.host_behavior_data = None
        cross_namespace_copy = self.ocfs & 0xc
        if cross_namespace_copy:
            # get host behavior support data
            get_features_cmd = f"{self.nvme_bin} get-feature {self.ctrl} " + \
                "--feature-id=0x16 --data-len=512 --raw-binary"
            result = self.run_cmd(get_features_cmd)
            err = result.returncode
            self.assertEqual(err, 0, "ERROR : nvme get-feature failed")
            self.host_behavior_data = result.stdout
            # enable cross-namespace copy formats
            if int.from_bytes(base64.b64decode(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] + str(cross_namespace_copy.to_bytes(2, 'little')) + self.host_behavior_data[6:]
                set_features_cmd = f"{self.nvme_bin} set-feature " + \
                    f"{self.ctrl} --feature-id=0x16 --data-len=512"
                result = self.run_cmd(set_features_cmd, stdin_data=data)
                self.assertEqual(result.returncode, 0, "Failed to enable cross-namespace copy formats")
        get_ns_id_cmd = f"{self.nvme_bin} get-ns-id {self.ns1}"
        result = self.run_cmd(get_ns_id_cmd)
        err = result.returncode
        self.assertEqual(err, 0, "ERROR : nvme get-ns-id failed")
        output = result.stdout
        self.ns1_nsid = int(output.strip().split(':')[-1])
        self.setup_log_dir(self.__class__.__name__)

    def tearDown(self):
        """ Post Section for TestNVMeCopy """
        if self.host_behavior_data:
            # restore saved host behavior support data
            set_features_cmd = f"{self.nvme_bin} set-feature {self.ctrl} " + \
                "--feature-id=0x16 --data-len=512"
            self.run_cmd(set_features_cmd, stdin_data=self.host_behavior_data)
        super().tearDown()

    def copy(self, sdlba, blocks, slbs, **kwargs):
        """ Wrapper for nvme copy
            - Args:
                - 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:
                - None
        """
        # 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"{self.nvme_bin} copy {self.ns1} " + \
            f"--format={desc_format} --sdlba={sdlba} --blocks={blocks} " + \
            f"--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
        self.assertEqual(self.exec_cmd(copy_cmd), 0)

    def test_copy(self):
        """ Testcase main """
        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)