310 lines · python
1#!/usr/bin/env python32# SPDX-License-Identifier: GPL-2.03 4from subprocess import PIPE, Popen5import json6import time7import argparse8import collections9import sys10 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 split17# to all valid configurations (e.g., split to 2, split to 4 etc.)18#19 20 21# Kselftest framework requirement - SKIP code is 422KSFT_SKIP=423Port = collections.namedtuple('Port', 'bus_info name')24 25 26def 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, stderr41 42 43class 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 arr70 71 def __init__(self, dev):72 self.if_names = devlink_ports.get_if_names(dev)73 74 75def 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" % port82 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 = 090 return lanes91 92 93def 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.name100 stdout, stderr = run_command(cmd)101 assert stderr == ""102 values = list(json.loads(stdout)['port'].values())[0]103 104 return values['splittable']105 106 107def 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 130def unsplit(port):131 """132 Unsplit $port.133 """134 135 cmd = "devlink port unsplit %s" % port136 stdout, stderr = run_command(cmd)137 test(stderr == "", "Unsplit port %s" % port)138 139 140def 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 == port147 for dev_port in devlink_ports.get_if_names(dev))148 149 150def exists_and_lanes(ports, lanes, dev):151 """152 Check if every port in the list $ports exists in the devlink ports and has153 $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 False162 if max_lanes != lanes:163 print("port %s has %d lanes, but %s were expected"164 % (port, lanes, max_lanes))165 return False166 return True167 168 169def 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 cond181 182 183def 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 192def split_unsplittable_port(port, k):193 """194 Test that splitting of unsplittable port fails.195 """196 197 # split to max198 new_split_group = split(k, port, should_fail=True)199 200 if new_split_group != []:201 unsplit(port.bus_info)202 203 204def 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 udev213 # 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 225def validate_devlink_output(devlink_data, target_property=None):226 """227 Determine if test should be skipped by checking:228 1. devlink_data contains values229 2. The target_property exist in devlink_data230 """231 skip_reason = None232 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 = None238 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 246def 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 parser254 255 256def main(cmdline=None):257 parser = make_parser()258 args = parser.parse_args(cmdline)259 260 dev = args.dev261 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" % dev271 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 = False279 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 all283 if max_lanes == 0:284 continue285 286 # If 1 lane, shouldn't be able to split287 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_lanes295 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 //= 2301 found_max_lanes = True302 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 308if __name__ == "__main__":309 main()310