Setup of VLAN interface

detd/ip.py

@@ -14,6 +14,8 @@
 
 
 import subprocess
+from pyroute2 import IPRoute
+from pyroute2.protocols import ETH_P_8021Q
 
 from .common import CommandString
 
@@ -26,77 +28,81 @@ def __init__(self):
         pass
 
 
-    def run(self, command):
-        cmd = command.split()
-        result = subprocess.run(cmd, capture_output=True, text=True)
+    def set_vlan(self, interface, stream, mapping):
 
-        success_codes = [0]
-        if result.returncode not in success_codes:
-            raise subprocess.CalledProcessError(result.returncode, command, result.stdout, result.stderr)
+        soprio_to_pcp = transform_soprio_to_pcp(mapping.soprio_to_pcp)
 
-        return result
+        name = "{}.{}".format(interface.name, stream.vid)
 
+        parent_interface_index = get_interface_index(interface.name)
 
-    def set_vlan(self, interface, stream, mapping):
+        if parent_interface_index is None:
+            raise ValueError("Interface {} could not be found".format(interface.name))
 
-        soprio_to_pcp = transform_soprio_to_pcp(mapping.soprio_to_pcp)
-        cmd = CommandStringIpLinkSetVlan(interface.name, stream.vid, soprio_to_pcp)
+        link_info = get_link_info(name)
+        if link_info:
+            # VLAN interface already exists, check for incompatible configuration
+            if get_ip_attr(link_info, 'IFLA_INFO_KIND') != 'vlan':
+                raise Exception("Existing interface {} has no VLAN link info".format(name))
 
-        self.run(str(cmd))
+            info_data = get_ip_attr(link_info, 'IFLA_INFO_DATA')
 
+            if get_ip_attr(info_data, 'IFLA_VLAN_PROTOCOL') != ETH_P_8021Q:
+                raise Exception("Existing interface {} is not a 802.1Q VLAN interface".format(name))
 
-    def unset_vlan(self, interface, stream):
-        cmd = CommandStringIpLinkUnsetVlan(interface.name, stream.vid)
+            if get_ip_attr(info_data, 'IFLA_VLAN_ID') != stream.vid:
+                raise Exception("Existing interface {} does not have VLAN ID {}".format(name, stream.vid))
+
+        ip = IPRoute()
+        ip.link('set' if link_info else 'add',
+                ifname = name,
+                kind = "vlan",
+                link = parent_interface_index,
+                vlan_id = stream.vid,
+                protocol = ETH_P_8021Q,
+                vlan_egress_qos = soprio_to_pcp
+                )
 
-        self.run(str(cmd))
+
+    def unset_vlan(self, interface, stream):
+        name = "{}.{}".format(interface.name, stream.vid)
+        ip = IPRoute()
+        ip.link('delete', ifname=name)
 
 
 def transform_soprio_to_pcp(soprio_to_pcp):
     mapping = []
     for soprio, pcp in soprio_to_pcp.items():
-        mapping.append("{0}:{1}".format(soprio, pcp))
-
-    return ' '.join(mapping)
-
-
-
-###############################################################################
-# ip command strings                                                          #
-###############################################################################
+        mapping.append(('IFLA_VLAN_QOS_MAPPING', {'from': soprio, 'to': pcp}))
 
-class CommandStringIpLinkSetVlan (CommandString):
+    return {'attrs': mapping}
 
-    def __init__(self, device, vid, soprio_to_pcp):
 
-        template = '''
-            ip link add
-                    link     $device
-                    name     $device.$id
-                    type     vlan
-                    protocol 802.1Q
-                    id       $id
-                    egress   $soprio_to_pcp'''
+def get_interface_index(name):
+    ip = IPRoute()
+    interface_index = ip.link_lookup(ifname=name)
 
-        params = {
-            'device'        : device,
-            'id'            : vid,
-            'soprio_to_pcp' : soprio_to_pcp
-        }
+    if not interface_index:
+        return None
 
-        super().__init__(template, params)
+    return interface_index[0]
 
 
+def get_link_info(interface):
+    index = get_interface_index(interface)
 
+    if index is None:
+        return None
 
-class CommandStringIpLinkUnsetVlan (CommandString):
+    ip = IPRoute()
+    for link in ip.get_links(index):
+        return get_ip_attr(link, 'IFLA_LINKINFO')
 
-    def __init__(self, device, vid):
 
-        template = 'ip link delete $device.$id'
+def get_ip_attr(data, key):
+    if any((subdata := attr)[0] == key for attr in data['attrs']):
+        return subdata[1]
+    else:
+        raise KeyError("Key {} not found".format(key))
 
-        params = {
-            'device' : device,
-            'id'     : vid
-        }
 
-        super().__init__(template, params)

detd/service.py

@@ -248,7 +248,7 @@ def _add_talker(self, request):
     def _mock_add_talker(self, request):
 
         with mock.patch.object(QdiscConfigurator,  'setup', return_value=None), \
-             mock.patch.object(CommandIp,   'run', return_value=None), \
+             mock.patch.object(CommandIp,   'set_vlan', return_value=None), \
              mock.patch.object(DeviceConfigurator, 'setup', return_value=None), \
              mock.patch.object(SystemInformation,  'get_pci_id', return_value=('8086:4B30')), \
              mock.patch.object(SystemInformation,  'get_rate', return_value=1000 * 1000 * 1000), \

setup.cfg

@@ -24,6 +24,7 @@ packages = find:
 python_requires = >=3.8, <3.9
 install_requires =
        protobuf ==3.6.1.3
+       pyroute2
 scripts =
 	setup_qos.sh
 	detd/detd

tests/common.py

@@ -49,7 +49,7 @@ def __init__(self, mode, qdisc_exc=None, vlan_exc=None, device_exc=None):
 
         if mode == TestMode.HOST:
             self.qdisc_conf_mock  = mock.patch.object(CommandTc,  'run', side_effect=qdisc_exc)
-            self.vlan_conf_mock   = mock.patch.object(CommandIp,   'run', side_effect=vlan_exc)
+            self.vlan_conf_mock   = mock.patch.object(CommandIp,   'set_vlan', side_effect=vlan_exc)
             self.device_conf_mock = mock.patch.object(CommandEthtool, 'run', side_effect=device_exc)
             self.device_pci_id_mock = mock.patch.object(SystemInformation, 'get_pci_id', return_value=('8086:4B30'))
             self.device_channels_mock = mock.patch.object(SystemInformation, 'get_channels_information', return_value=(8,8))

tests/test_commandstring.py

@@ -8,8 +8,6 @@
 import re
 import unittest
 
-from detd import CommandStringIpLinkSetVlan
-from detd import CommandStringIpLinkUnsetVlan
 from detd import CommandStringEthtoolFeatures
 from detd import CommandStringEthtoolGetChannelsInformation
 from detd import CommandStringEthtoolSetCombinedChannels
@@ -32,42 +30,6 @@ def assert_commandstring_equal(self, one, another):
         self.assertEqual(harmonized_one, harmonized_another)
 
 
-
-
-    def test_iplinksetvlan(self):
-
-        interface_name = "eth0"
-        stream_vid = 3
-        soprio_to_pcp = "0:7 1:6 2:5 3:4 4:3 5:2 6:1 7:0"
-
-        cmd = CommandStringIpLinkSetVlan(interface_name, stream_vid, soprio_to_pcp)
-        expected = """
-            ip link add
-                    link     eth0
-                    name     eth0.3
-                    type     vlan
-                    protocol 802.1Q
-                    id       3
-                    egress   0:7 1:6 2:5 3:4 4:3 5:2 6:1 7:0"""
-
-        self.assert_commandstring_equal(cmd, expected)
-
-
-
-
-    def test_iplinkunsetvlan(self):
-
-        interface_name = "eth0"
-        stream_vid = 3
-
-        cmd = CommandStringIpLinkUnsetVlan(interface_name, stream_vid)
-        expected = "ip link delete eth0.3"
-
-        self.assert_commandstring_equal(cmd, expected)
-
-
-
-
     def test_ethtoolfeatures(self):
 
         interface_name = "eth0"

tests/test_manager.py

@@ -232,7 +232,7 @@ def test_add_talker_vlan_error(self):
 
         config = setup_config(self.mode)
 
-        with RunContext(self.mode, vlan_exc=subprocess.CalledProcessError(1, "ip")):
+        with RunContext(self.mode, vlan_exc=ValueError("Interface could not be found")):
             manager = Manager()
 
             self.assertRaises(RuntimeError, manager.add_talker, config)