bridge: T3042: Support VLAN filter and VLAN sub-interface on the bridge

1 files changed, 92 insertions, 0 deletions

diff --git a/smoketest/scripts/cli/test_interfaces_bridge.py b/smoketest/scripts/cli/test_interfaces_bridge.py
index a1359680b..3b7f1bc9a 100755
--- a/smoketest/scripts/cli/test_interfaces_bridge.py
+++ b/smoketest/scripts/cli/test_interfaces_bridge.py
@@ -21,12 +21,16 @@ from base_interfaces_test import BasicInterfaceTest
 from glob import glob
 from netifaces import interfaces
 from vyos.ifconfig import Section
+from vyos.util import cmd
+import json
 
 class BridgeInterfaceTest(BasicInterfaceTest.BaseTest):
     def setUp(self):
         super().setUp()
 
         self._test_ipv6 = True
+        self._test_vlan = True
+        self._test_qinq = True
 
         self._base_path = ['interfaces', 'bridge']
         self._interfaces = ['br0']
@@ -78,6 +82,94 @@ class BridgeInterfaceTest(BasicInterfaceTest.BaseTest):
             self.session.delete(self._base_path + [interface, 'member'])
 
         self.session.commit()
+    
+    def test_vlan_filter(self):
+        """ Add member interface to bridge and set VLAN filter """
+        for interface in self._interfaces:
+            base = self._base_path + [interface]
+            self.session.set(base + ['address', '192.0.2.1/24'])
+            self.session.set(base + ['vif', '2','address','192.0.3.1/24'])
+
+            vlan_id = 101
+            allowed_vlan = 2
+            allowed_vlan_range = '4-9'
+            # assign members to bridge interface
+            for member in self._members:
+                base_member = base + ['member', 'interface', member]
+                self.session.set(base_member + ['allowed-vlan', str(allowed_vlan)])
+                self.session.set(base_member + ['allowed-vlan', allowed_vlan_range])
+                self.session.set(base_member + ['native-vlan', str(vlan_id)])
+                vlan_id += 1
+
+        # commit config
+        self.session.commit()
+        
+        # Detect the vlan filter function
+        for interface in self._interfaces:
+            with open(f'/sys/class/net/{interface}/bridge/vlan_filtering', 'r') as f:
+                flags = f.read()
+                self.assertEqual(int(flags), 1)
+        
+        # Execute the program to obtain status information
+        
+        json_data = cmd('bridge -j vlan show', shell=True)
+        
+        vlan_filter_status = None
+        
+        vlan_filter_status = json.loads(json_data)
+            
+        
+        if vlan_filter_status is not None:
+            for interface_status in vlan_filter_status:
+                ifname = interface_status['ifname']
+                for interface in self._members:
+                    vlan_success = 0;
+                    if interface == ifname:
+                        vlans_status = interface_status['vlans']
+                        for vlan_status in vlans_status:
+                            vlan_id = vlan_status['vlan']
+                            flag_num = 0
+                            if 'flags' in vlan_status:
+                                flags = vlan_status['flags']
+                                for flag in flags:
+                                    flag_num = flag_num +1
+                            if vlan_id == 2:
+                                if flag_num == 0:
+                                    vlan_success = vlan_success + 1
+                            else:
+                                for id in range(4,10):
+                                    if vlan_id == id:
+                                        if flag_num == 0:
+                                            vlan_success = vlan_success + 1
+                                if vlan_id >= 101:
+                                    if flag_num == 2:
+                                        vlan_success = vlan_success + 1
+                        if vlan_success >= 7:
+                            self.assertTrue(True)
+                        else:
+                            self.assertTrue(False)
+                        
+        else:
+            self.assertTrue(False)
+        
+            
+        
+
+        # check member interfaces are added on the bridge
+        
+        for interface in self._interfaces:
+            bridge_members = []
+            for tmp in glob(f'/sys/class/net/{interface}/lower_*'):
+                bridge_members.append(os.path.basename(tmp).replace('lower_', ''))
+
+            for member in self._members:
+                self.assertIn(member, bridge_members)
+
+        # delete all members
+        for interface in self._interfaces:
+            self.session.delete(self._base_path + [interface, 'member'])
+
+        self.session.commit()
 
     def test_vlan_members(self):
         """ T2945: ensure that VIFs are not dropped from bridge """