tools/testing/selftests/drivers/net/hw/devlink_port_split.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: GPL-2.0
   3
   4 from subprocess import PIPE, Popen
   5 import json
   6 import time
   7 import argparse
   8 import collections
   9 import sys
  10
  11 #
  12 # Test port split configuration using devlink-port lanes attribute.
  13 # The test is skipped in case the attribute is not available.
  14 #
  15 # First, check that all the ports with 1 lane fail to split.
  16 # Second, check that all the ports with more than 1 lane can be split
  17 # to all valid configurations (e.g., split to 2, split to 4 etc.)
  18 #
  19
  20
  21 # Kselftest framework requirement - SKIP code is 4
  22 KSFT_SKIP=4
  23 Port = collections.namedtuple('Port', 'bus_info name')
  24
  25
  26 def run_command(cmd, should_fail=False):
  27     """
  28     Run a command in subprocess.
  29     Return: Tuple of (stdout, stderr).
  30     """
  31
  32     p = Popen(cmd, stdout=PIPE, stderr=PIPE, shell=True)
  33     stdout, stderr = p.communicate()
  34     stdout, stderr = stdout.decode(), stderr.decode()
  35
  36     if stderr != "" and not should_fail:
  37         print("Error sending command: %s" % cmd)
  38         print(stdout)
  39         print(stderr)
  40     return stdout, stderr
  41
  42
  43 class devlink_ports(object):
  44     """
  45     Class that holds information on the devlink ports, required to the tests;
  46     if_names: A list of interfaces in the devlink ports.
  47     """
  48
  49     def get_if_names(dev):
  50         """
  51         Get a list of physical devlink ports.
  52         Return: Array of tuples (bus_info/port, if_name).
  53         """
  54
  55         arr = []
  56
  57         cmd = "devlink -j port show"
  58         stdout, stderr = run_command(cmd)
  59         assert stderr == ""
  60         ports = json.loads(stdout)['port']
  61
  62         validate_devlink_output(ports, 'flavour')
  63
  64         for port in ports:
  65             if dev in port:
  66                 if ports[port]['flavour'] == 'physical':
  67                     arr.append(Port(bus_info=port, name=ports[port]['netdev']))
  68
  69         return arr
  70
  71     def __init__(self, dev):
  72         self.if_names = devlink_ports.get_if_names(dev)
  73
  74
  75 def get_max_lanes(port):
  76     """
  77     Get the $port's maximum number of lanes.
  78     Return: number of lanes, e.g. 1, 2, 4 and 8.
  79     """
  80
  81     cmd = "devlink -j port show %s" % port
  82     stdout, stderr = run_command(cmd)
  83     assert stderr == ""
  84     values = list(json.loads(stdout)['port'].values())[0]
  85
  86     if 'lanes' in values:
  87         lanes = values['lanes']
  88     else:
  89         lanes = 0
  90     return lanes
  91
  92
  93 def get_split_ability(port):
  94     """
  95     Get the $port split ability.
  96     Return: split ability, true or false.
  97     """
  98
  99     cmd = "devlink -j port show %s" % port.name
 100     stdout, stderr = run_command(cmd)
 101     assert stderr == ""
 102     values = list(json.loads(stdout)['port'].values())[0]
 103
 104     return values['splittable']
 105
 106
 107 def split(k, port, should_fail=False):
 108     """
 109     Split $port into $k ports.
 110     If should_fail == True, the split should fail. Otherwise, should pass.
 111     Return: Array of sub ports after splitting.
 112             If the $port wasn't split, the array will be empty.
 113     """
 114
 115     cmd = "devlink port split %s count %s" % (port.bus_info, k)
 116     stdout, stderr = run_command(cmd, should_fail=should_fail)
 117
 118     if should_fail:
 119         if not test(stderr != "", "%s is unsplittable" % port.name):
 120             print("split an unsplittable port %s" % port.name)
 121             return create_split_group(port, k)
 122     else:
 123         if stderr == "":
 124             return create_split_group(port, k)
 125         print("didn't split a splittable port %s" % port.name)
 126
 127     return []
 128
 129
 130 def unsplit(port):
 131     """
 132     Unsplit $port.
 133     """
 134
 135     cmd = "devlink port unsplit %s" % port
 136     stdout, stderr = run_command(cmd)
 137     test(stderr == "", "Unsplit port %s" % port)
 138
 139
 140 def exists(port, dev):
 141     """
 142     Check if $port exists in the devlink ports.
 143     Return: True is so, False otherwise.
 144     """
 145
 146     return any(dev_port.name == port
 147                for dev_port in devlink_ports.get_if_names(dev))
 148
 149
 150 def exists_and_lanes(ports, lanes, dev):
 151     """
 152     Check if every port in the list $ports exists in the devlink ports and has
 153     $lanes number of lanes after splitting.
 154     Return: True if both are True, False otherwise.
 155     """
 156
 157     for port in ports:
 158         max_lanes = get_max_lanes(port)
 159         if not exists(port, dev):
 160             print("port %s doesn't exist in devlink ports" % port)
 161             return False
 162         if max_lanes != lanes:
 163             print("port %s has %d lanes, but %s were expected"
 164                   % (port, lanes, max_lanes))
 165             return False
 166     return True
 167
 168
 169 def test(cond, msg):
 170     """
 171     Check $cond and print a message accordingly.
 172     Return: True is pass, False otherwise.
 173     """
 174
 175     if cond:
 176         print("TEST: %-60s [ OK ]" % msg)
 177     else:
 178         print("TEST: %-60s [FAIL]" % msg)
 179
 180     return cond
 181
 182
 183 def create_split_group(port, k):
 184     """
 185     Create the split group for $port.
 186     Return: Array with $k elements, which are the split port group.
 187     """
 188
 189     return list(port.name + "s" + str(i) for i in range(k))
 190
 191
 192 def split_unsplittable_port(port, k):
 193     """
 194     Test that splitting of unsplittable port fails.
 195     """
 196
 197     # split to max
 198     new_split_group = split(k, port, should_fail=True)
 199
 200     if new_split_group != []:
 201         unsplit(port.bus_info)
 202
 203
 204 def split_splittable_port(port, k, lanes, dev):
 205     """
 206     Test that splitting of splittable port passes correctly.
 207     """
 208
 209     new_split_group = split(k, port)
 210
 211     # Once the split command ends, it takes some time to the sub ifaces'
 212     # to get their names. Use udevadm to continue only when all current udev
 213     # events are handled.
 214     cmd = "udevadm settle"
 215     stdout, stderr = run_command(cmd)
 216     assert stderr == ""
 217
 218     if new_split_group != []:
 219         test(exists_and_lanes(new_split_group, lanes/k, dev),
 220              "split port %s into %s" % (port.name, k))
 221
 222     unsplit(port.bus_info)
 223
 224
 225 def validate_devlink_output(devlink_data, target_property=None):
 226     """
 227     Determine if test should be skipped by checking:
 228       1. devlink_data contains values
 229       2. The target_property exist in devlink_data
 230     """
 231     skip_reason = None
 232     if any(devlink_data.values()):
 233         if target_property:
 234             skip_reason = "{} not found in devlink output, test skipped".format(target_property)
 235             for key in devlink_data:
 236                 if target_property in devlink_data[key]:
 237                     skip_reason = None
 238     else:
 239         skip_reason = 'devlink output is empty, test skipped'
 240
 241     if skip_reason:
 242         print(skip_reason)
 243         sys.exit(KSFT_SKIP)
 244
 245
 246 def make_parser():
 247     parser = argparse.ArgumentParser(description='A test for port splitting.')
 248     parser.add_argument('--dev',
 249                         help='The devlink handle of the device under test. ' +
 250                              'The default is the first registered devlink ' +
 251                              'handle.')
 252
 253     return parser
 254
 255
 256 def main(cmdline=None):
 257     parser = make_parser()
 258     args = parser.parse_args(cmdline)
 259
 260     dev = args.dev
 261     if not dev:
 262         cmd = "devlink -j dev show"
 263         stdout, stderr = run_command(cmd)
 264         assert stderr == ""
 265
 266         validate_devlink_output(json.loads(stdout))
 267         devs = json.loads(stdout)['dev']
 268         dev = list(devs.keys())[0]
 269
 270     cmd = "devlink dev show %s" % dev
 271     stdout, stderr = run_command(cmd)
 272     if stderr != "":
 273         print("devlink device %s can not be found" % dev)
 274         sys.exit(1)
 275
 276     ports = devlink_ports(dev)
 277
 278     found_max_lanes = False
 279     for port in ports.if_names:
 280         max_lanes = get_max_lanes(port.name)
 281
 282         # If max lanes is 0, do not test port splitting at all
 283         if max_lanes == 0:
 284             continue
 285
 286         # If 1 lane, shouldn't be able to split
 287         elif max_lanes == 1:
 288             test(not get_split_ability(port),
 289                  "%s should not be able to split" % port.name)
 290             split_unsplittable_port(port, max_lanes)
 291
 292         # Else, splitting should pass and all the split ports should exist.
 293         else:
 294             lane = max_lanes
 295             test(get_split_ability(port),
 296                  "%s should be able to split" % port.name)
 297             while lane > 1:
 298                 split_splittable_port(port, lane, max_lanes, dev)
 299
 300                 lane //= 2
 301         found_max_lanes = True
 302
 303     if not found_max_lanes:
 304         print(f"Test not started, no port of device {dev} reports max_lanes")
 305         sys.exit(KSFT_SKIP)
 306
 307
 308 if __name__ == "__main__":
 309     main()