• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# SPDX-License-Identifier: GPL-2.0
3
4from subprocess import PIPE, Popen
5import json
6import time
7import argparse
8import collections
9import 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
22KSFT_SKIP=4
23Port = 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, stderr
41
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 arr
70
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" % 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
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.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
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" % port
136    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 == port
147               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 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
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 cond
181
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 max
198    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 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
225def 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
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 parser
254
255
256def 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
308if __name__ == "__main__":
309    main()
310