smoketest: mirror: T3169: re-add mirror / SPAN test case

Commit ef629504d4 ("smoketest: mirror: drop testcase") dropped the smoketests
entirely as they have been programmed in a wrong way leading to outages on live
development systems which used dummy interfaces for management traffic.

11 files changed, 77 insertions, 45 deletions

diff --git a/smoketest/scripts/cli/base_interfaces_test.py b/smoketest/scripts/cli/base_interfaces_test.py
index a784140f3..8ee5395d0 100644
--- a/smoketest/scripts/cli/base_interfaces_test.py
+++ b/smoketest/scripts/cli/base_interfaces_test.py
@@ -25,6 +25,7 @@ from netifaces import AF_INET6
 
 from vyos.configsession import ConfigSession
 from vyos.ifconfig import Interface
+from vyos.ifconfig import Section
 from vyos.util import read_file
 from vyos.util import cmd
 from vyos.util import dict_search
@@ -32,6 +33,25 @@ from vyos.util import process_named_running
 from vyos.validate import is_intf_addr_assigned
 from vyos.validate import is_ipv6_link_local
 
+def is_mirrored_to(interface, mirror_if, qdisc):
+    """
+    Ask TC if we are mirroring traffic to a discrete interface.
+
+    interface: source interface
+    mirror_if: destination where we mirror our data to
+    qdisc: must be ffff or 1 for ingress/egress
+    """
+    if qdisc not in ['ffff', '1']:
+        raise ValueError()
+
+    ret_val = False
+    tmp = cmd(f'tc -s -p filter ls dev {interface} parent {qdisc}: | grep mirred')
+    tmp = tmp.lower()
+    if mirror_if in tmp:
+        ret_val = True
+    return ret_val
+
+
 dhcp6c_config_file = '/run/dhcp6c/dhcp6c.{}.conf'
 def get_dhcp6c_config_value(interface, key):
     tmp = read_file(dhcp6c_config_file.format(interface))
@@ -56,25 +76,53 @@ class BasicInterfaceTest:
         _interfaces = []
         _qinq_range = ['10', '20', '30']
         _vlan_range = ['100', '200', '300', '2000']
+        _test_addr = ['192.0.2.1/26', '192.0.2.255/31', '192.0.2.64/32',
+                      '2001:db8:1::ffff/64', '2001:db8:101::1/112']
+
+        _mirror_interfaces = []
         # choose IPv6 minimum MTU value for tests - this must always work
         _mtu = '1280'
 
         def setUp(self):
             self.session = ConfigSession(os.getpid())
 
-            self._test_addr = ['192.0.2.1/26', '192.0.2.255/31', '192.0.2.64/32',
-                                '2001:db8:1::ffff/64', '2001:db8:101::1/112']
-            self._test_mtu = False
-            self._options = {}
+            # Setup mirror interfaces for SPAN (Switch Port Analyzer)
+            for span in self._mirror_interfaces:
+                section = Section.section(span)
+                self.session.set(['interfaces', section, span])
 
         def tearDown(self):
             # Ethernet is handled in its derived class
             if 'ethernet' not in self._base_path:
                 self.session.delete(self._base_path)
 
+            # Tear down mirror interfaces for SPAN (Switch Port Analyzer)
+            for span in self._mirror_interfaces:
+                section = Section.section(span)
+                self.session.delete(['interfaces', section, span])
+
             self.session.commit()
             del self.session
 
+        def test_span_mirror(self):
+            if not self._mirror_interfaces:
+                return None
+
+            # Check the two-way mirror rules of ingress and egress
+            for mirror in self._mirror_interfaces:
+                for interface in self._interfaces:
+                    self.session.set(self._base_path + [interface, 'mirror', 'ingress', mirror])
+                    self.session.set(self._base_path + [interface, 'mirror', 'egress',  mirror])
+
+            self.session.commit()
+
+            # Verify config
+            for mirror in self._mirror_interfaces:
+                for interface in self._interfaces:
+                    self.assertTrue(is_mirrored_to(interface, mirror, 'ffff'))
+                    self.assertTrue(is_mirrored_to(interface, mirror, '1'))
+
+
         def test_interface_description(self):
             # Check if description can be added to interface and
             # can be read back
diff --git a/smoketest/scripts/cli/test_interfaces_bonding.py b/smoketest/scripts/cli/test_interfaces_bonding.py
index d38e11a63..a35682b7c 100755
--- a/smoketest/scripts/cli/test_interfaces_bonding.py
+++ b/smoketest/scripts/cli/test_interfaces_bonding.py
@@ -26,17 +26,15 @@ from vyos.util import read_file
 
 class BondingInterfaceTest(BasicInterfaceTest.BaseTest):
     def setUp(self):
-        super().setUp()
-
-        self._base_path = ['interfaces', 'bonding']
-        self._interfaces = ['bond0']
         self._test_mtu = True
         self._test_vlan = True
         self._test_qinq = True
         self._test_ipv6 = True
-        self._test_mirror = True
-
+        self._base_path = ['interfaces', 'bonding']
+        self._interfaces = ['bond0']
+        self._mirror_interfaces = ['dum21354']
         self._members = []
+
         # we need to filter out VLAN interfaces identified by a dot (.)
         # in their name - just in case!
         if 'TEST_ETH' in os.environ:
@@ -50,6 +48,8 @@ class BondingInterfaceTest(BasicInterfaceTest.BaseTest):
         for member in self._members:
             self._options['bond0'].append(f'member interface {member}')
 
+        super().setUp()
+
 
     def test_add_single_ip_address(self):
         super().test_add_single_ip_address()
diff --git a/smoketest/scripts/cli/test_interfaces_bridge.py b/smoketest/scripts/cli/test_interfaces_bridge.py
index 394d50025..7444701c1 100755
--- a/smoketest/scripts/cli/test_interfaces_bridge.py
+++ b/smoketest/scripts/cli/test_interfaces_bridge.py
@@ -28,17 +28,13 @@ from vyos.util import read_file
 
 class BridgeInterfaceTest(BasicInterfaceTest.BaseTest):
     def setUp(self):
-        super().setUp()
-
         self._test_ipv6 = True
         self._test_vlan = True
         self._test_qinq = True
-        self._test_mirror = True
-
         self._base_path = ['interfaces', 'bridge']
-        self._interfaces = ['br0']
-
+        self._mirror_interfaces = ['dum21354']
         self._members = []
+
         # we need to filter out VLAN interfaces identified by a dot (.)
         # in their name - just in case!
         if 'TEST_ETH' in os.environ:
@@ -51,7 +47,9 @@ class BridgeInterfaceTest(BasicInterfaceTest.BaseTest):
         self._options['br0'] = []
         for member in self._members:
             self._options['br0'].append(f'member interface {member}')
+        self._interfaces = list(self._options)
 
+        super().setUp()
 
     def test_add_remove_bridge_member(self):
         # Add member interfaces to bridge and set STP cost/priority
@@ -188,5 +186,5 @@ class BridgeInterfaceTest(BasicInterfaceTest.BaseTest):
                 self.session.delete(['interfaces', 'ethernet', member, 'vif', vif])
 
 if __name__ == '__main__':
-    unittest.main(verbosity=2)
+    unittest.main(verbosity=2, failfast=True)
 
diff --git a/smoketest/scripts/cli/test_interfaces_dummy.py b/smoketest/scripts/cli/test_interfaces_dummy.py
index 97f5344ac..c482a6f0b 100755
--- a/smoketest/scripts/cli/test_interfaces_dummy.py
+++ b/smoketest/scripts/cli/test_interfaces_dummy.py
@@ -20,9 +20,9 @@ from base_interfaces_test import BasicInterfaceTest
 
 class DummyInterfaceTest(BasicInterfaceTest.BaseTest):
     def setUp(self):
-         super().setUp()
          self._base_path = ['interfaces', 'dummy']
          self._interfaces = ['dum0', 'dum1', 'dum2']
+         super().setUp()
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_interfaces_ethernet.py b/smoketest/scripts/cli/test_interfaces_ethernet.py
index 2d0a4827d..3c4796283 100755
--- a/smoketest/scripts/cli/test_interfaces_ethernet.py
+++ b/smoketest/scripts/cli/test_interfaces_ethernet.py
@@ -35,16 +35,13 @@ def get_wpa_supplicant_value(interface, key):
 
 class EthernetInterfaceTest(BasicInterfaceTest.BaseTest):
     def setUp(self):
-        super().setUp()
-
-        self._base_path = ['interfaces', 'ethernet']
         self._test_ip = True
         self._test_mtu = True
         self._test_vlan = True
         self._test_qinq = True
         self._test_ipv6 = True
-        self._test_mirror = True
-        self._interfaces = []
+        self._base_path = ['interfaces', 'ethernet']
+        self._mirror_interfaces = ['dum21354']
 
         # we need to filter out VLAN interfaces identified by a dot (.)
         # in their name - just in case!
@@ -66,6 +63,8 @@ class EthernetInterfaceTest(BasicInterfaceTest.BaseTest):
                 mac = read_file(f'/sys/class/net/{interface}/address')
             self._macs[interface] = mac
 
+        super().setUp()
+
 
     def tearDown(self):
         for interface in self._interfaces:
diff --git a/smoketest/scripts/cli/test_interfaces_geneve.py b/smoketest/scripts/cli/test_interfaces_geneve.py
index 7e0389a63..98f55210f 100755
--- a/smoketest/scripts/cli/test_interfaces_geneve.py
+++ b/smoketest/scripts/cli/test_interfaces_geneve.py
@@ -14,24 +14,20 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import os
 import unittest
 
-from vyos.configsession import ConfigSession, ConfigSessionError
+from vyos.configsession import ConfigSession
 from base_interfaces_test import BasicInterfaceTest
 
-
 class GeneveInterfaceTest(BasicInterfaceTest.BaseTest):
     def setUp(self):
-        super().setUp()
-
         self._base_path = ['interfaces', 'geneve']
         self._options = {
             'gnv0': ['vni 10', 'remote 127.0.1.1'],
             'gnv1': ['vni 20', 'remote 127.0.1.2'],
         }
         self._interfaces = list(self._options)
-
+        super().setUp()
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_interfaces_l2tpv3.py b/smoketest/scripts/cli/test_interfaces_l2tpv3.py
index be9565d00..c756bfdd5 100755
--- a/smoketest/scripts/cli/test_interfaces_l2tpv3.py
+++ b/smoketest/scripts/cli/test_interfaces_l2tpv3.py
@@ -15,7 +15,6 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import json
-import jmespath
 import unittest
 
 from base_interfaces_test import BasicInterfaceTest
@@ -23,8 +22,6 @@ from vyos.util import cmd
 
 class GeneveInterfaceTest(BasicInterfaceTest.BaseTest):
     def setUp(self):
-        super().setUp()
-
         self._base_path = ['interfaces', 'l2tpv3']
         self._options = {
             'l2tpeth10': ['local-ip 127.0.0.1', 'remote-ip 127.10.10.10',
@@ -37,6 +34,7 @@ class GeneveInterfaceTest(BasicInterfaceTest.BaseTest):
                           'source-port 2020', 'destination-port 20202'],
         }
         self._interfaces = list(self._options)
+        super().setUp()
 
     def test_add_single_ip_address(self):
         super().test_add_single_ip_address()
diff --git a/smoketest/scripts/cli/test_interfaces_pseudo_ethernet.py b/smoketest/scripts/cli/test_interfaces_pseudo_ethernet.py
index c1711c5a3..85e5e70bd 100755
--- a/smoketest/scripts/cli/test_interfaces_pseudo_ethernet.py
+++ b/smoketest/scripts/cli/test_interfaces_pseudo_ethernet.py
@@ -19,22 +19,19 @@ import unittest
 from base_interfaces_test import BasicInterfaceTest
 
 class PEthInterfaceTest(BasicInterfaceTest.BaseTest):
-
     def setUp(self):
-        super().setUp()
-        self._base_path = ['interfaces', 'pseudo-ethernet']
-
         self._test_ip = True
         self._test_ipv6 = True
         self._test_mtu = True
         self._test_vlan = True
         self._test_qinq = True
-
+        self._base_path = ['interfaces', 'pseudo-ethernet']
         self._options = {
             'peth0': ['source-interface eth1'],
             'peth1': ['source-interface eth1'],
         }
         self._interfaces = list(self._options)
+        super().setUp()
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_interfaces_tunnel.py b/smoketest/scripts/cli/test_interfaces_tunnel.py
index 45679e280..f7b7f99ca 100755
--- a/smoketest/scripts/cli/test_interfaces_tunnel.py
+++ b/smoketest/scripts/cli/test_interfaces_tunnel.py
@@ -62,11 +62,8 @@ def tunnel_conf(interface):
 
 class TunnelInterfaceTest(BasicInterfaceTest.BaseTest):
     def setUp(self):
-        super().setUp()
-
-        self._base_path = ['interfaces', 'tunnel']
         self._test_mtu = True
-
+        self._base_path = ['interfaces', 'tunnel']
         self.local_v4 = '192.0.2.1'
         self.local_v6 = '2001:db8::1'
 
@@ -79,6 +76,7 @@ class TunnelInterfaceTest(BasicInterfaceTest.BaseTest):
         }
 
         self._interfaces = list(self._options)
+        super().setUp()
 
     def tearDown(self):
         self.session.delete(['interfaces', 'dummy', source_if])
diff --git a/smoketest/scripts/cli/test_interfaces_vxlan.py b/smoketest/scripts/cli/test_interfaces_vxlan.py
index f41c180ad..a9b0fc5a1 100755
--- a/smoketest/scripts/cli/test_interfaces_vxlan.py
+++ b/smoketest/scripts/cli/test_interfaces_vxlan.py
@@ -21,8 +21,6 @@ from base_interfaces_test import BasicInterfaceTest
 
 class VXLANInterfaceTest(BasicInterfaceTest.BaseTest):
     def setUp(self):
-        super().setUp()
-
         self._test_mtu = True
         self._base_path = ['interfaces', 'vxlan']
         self._options = {
@@ -30,6 +28,7 @@ class VXLANInterfaceTest(BasicInterfaceTest.BaseTest):
             'vxlan1': ['vni 20', 'group 239.1.1.1', 'source-interface eth0'],
         }
         self._interfaces = list(self._options)
+        super().setUp()
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_interfaces_wireless.py b/smoketest/scripts/cli/test_interfaces_wireless.py
index 9d2f4ea59..ffaa7d523 100755
--- a/smoketest/scripts/cli/test_interfaces_wireless.py
+++ b/smoketest/scripts/cli/test_interfaces_wireless.py
@@ -33,8 +33,6 @@ def get_config_value(interface, key):
 
 class WirelessInterfaceTest(BasicInterfaceTest.BaseTest):
     def setUp(self):
-        super().setUp()
-
         self._base_path = ['interfaces', 'wireless']
         self._options = {
             'wlan0':  ['physical-device phy0', 'ssid VyOS-WIFI-0',
@@ -47,6 +45,7 @@ class WirelessInterfaceTest(BasicInterfaceTest.BaseTest):
                        'type access-point', 'address 192.0.2.13/30', 'channel 0'],
         }
         self._interfaces = list(self._options)
+        super().setUp()
 
     def test_wireless_add_single_ip_address(self):
         # derived method to check if member interfaces are enslaved properly