2 # SPDX-License-Identifier: GPL-2.0
4 from subprocess import PIPE, Popen
12 # Test port split configuration using devlink-port lanes attribute.
13 # The test is skipped in case the attribute is not available.
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.)
21 # Kselftest framework requirement - SKIP code is 4
23 Port = collections.namedtuple('Port', 'bus_info name')
26 def run_command(cmd, should_fail=False):
28 Run a command in subprocess.
29 Return: Tuple of (stdout, stderr).
32 p = Popen(cmd, stdout=PIPE, stderr=PIPE, shell=True)
33 stdout, stderr = p.communicate()
34 stdout, stderr = stdout.decode(), stderr.decode()
36 if stderr != "" and not should_fail:
37 print("Error sending command: %s" % cmd)
43 class devlink_ports(object):
45 Class that holds information on the devlink ports, required to the tests;
46 if_names: A list of interfaces in the devlink ports.
49 def get_if_names(dev):
51 Get a list of physical devlink ports.
52 Return: Array of tuples (bus_info/port, if_name).
57 cmd = "devlink -j port show"
58 stdout, stderr = run_command(cmd)
60 ports = json.loads(stdout)['port']
62 validate_devlink_output(ports, 'flavour')
66 if ports[port]['flavour'] == 'physical':
67 arr.append(Port(bus_info=port, name=ports[port]['netdev']))
71 def __init__(self, dev):
72 self.if_names = devlink_ports.get_if_names(dev)
75 def get_max_lanes(port):
77 Get the $port's maximum number of lanes.
78 Return: number of lanes, e.g. 1, 2, 4 and 8.
81 cmd = "devlink -j port show %s" % port
82 stdout, stderr = run_command(cmd)
84 values = list(json.loads(stdout)['port'].values())[0]
87 lanes = values['lanes']
93 def get_split_ability(port):
95 Get the $port split ability.
96 Return: split ability, true or false.
99 cmd = "devlink -j port show %s" % port.name
100 stdout, stderr = run_command(cmd)
102 values = list(json.loads(stdout)['port'].values())[0]
104 return values['splittable']
107 def split(k, port, should_fail=False):
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.
115 cmd = "devlink port split %s count %s" % (port.bus_info, k)
116 stdout, stderr = run_command(cmd, should_fail=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)
124 return create_split_group(port, k)
125 print("didn't split a splittable port %s" % port.name)
135 cmd = "devlink port unsplit %s" % port
136 stdout, stderr = run_command(cmd)
137 test(stderr == "", "Unsplit port %s" % port)
140 def exists(port, dev):
142 Check if $port exists in the devlink ports.
143 Return: True is so, False otherwise.
146 return any(dev_port.name == port
147 for dev_port in devlink_ports.get_if_names(dev))
150 def exists_and_lanes(ports, lanes, dev):
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.
158 max_lanes = get_max_lanes(port)
159 if not exists(port, dev):
160 print("port %s doesn't exist in devlink ports" % port)
162 if max_lanes != lanes:
163 print("port %s has %d lanes, but %s were expected"
164 % (port, lanes, max_lanes))
171 Check $cond and print a message accordingly.
172 Return: True is pass, False otherwise.
176 print("TEST: %-60s [ OK ]" % msg)
178 print("TEST: %-60s [FAIL]" % msg)
183 def create_split_group(port, k):
185 Create the split group for $port.
186 Return: Array with $k elements, which are the split port group.
189 return list(port.name + "s" + str(i) for i in range(k))
192 def split_unsplittable_port(port, k):
194 Test that splitting of unsplittable port fails.
198 new_split_group = split(k, port, should_fail=True)
200 if new_split_group != []:
201 unsplit(port.bus_info)
204 def split_splittable_port(port, k, lanes, dev):
206 Test that splitting of splittable port passes correctly.
209 new_split_group = split(k, port)
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)
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))
222 unsplit(port.bus_info)
225 def validate_devlink_output(devlink_data, target_property=None):
227 Determine if test should be skipped by checking:
228 1. devlink_data contains values
229 2. The target_property exist in devlink_data
232 if any(devlink_data.values()):
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]:
239 skip_reason = 'devlink output is empty, test skipped'
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 ' +
256 def main(cmdline=None):
257 parser = make_parser()
258 args = parser.parse_args(cmdline)
262 cmd = "devlink -j dev show"
263 stdout, stderr = run_command(cmd)
266 validate_devlink_output(json.loads(stdout))
267 devs = json.loads(stdout)['dev']
268 dev = list(devs.keys())[0]
270 cmd = "devlink dev show %s" % dev
271 stdout, stderr = run_command(cmd)
273 print("devlink device %s can not be found" % dev)
276 ports = devlink_ports(dev)
278 found_max_lanes = False
279 for port in ports.if_names:
280 max_lanes = get_max_lanes(port.name)
282 # If max lanes is 0, do not test port splitting at all
286 # If 1 lane, shouldn't be able to split
288 test(not get_split_ability(port),
289 "%s should not be able to split" % port.name)
290 split_unsplittable_port(port, max_lanes)
292 # Else, splitting should pass and all the split ports should exist.
295 test(get_split_ability(port),
296 "%s should be able to split" % port.name)
298 split_splittable_port(port, lane, max_lanes, dev)
301 found_max_lanes = True
303 if not found_max_lanes:
304 print(f"Test not started, no port of device {dev} reports max_lanes")
308 if __name__ == "__main__":