diff options
Diffstat (limited to 'smoketest/scripts/cli/base_interfaces_test.py')
| -rw-r--r-- | smoketest/scripts/cli/base_interfaces_test.py | 56 |
1 files changed, 52 insertions, 4 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 |