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