diff options
Diffstat (limited to 'smoketest/scripts/cli/base_interfaces_test.py')
| -rw-r--r-- | smoketest/scripts/cli/base_interfaces_test.py | 139 |
1 files changed, 124 insertions, 15 deletions
diff --git a/smoketest/scripts/cli/base_interfaces_test.py b/smoketest/scripts/cli/base_interfaces_test.py index a3868fa70..b5b65e253 100644 --- a/smoketest/scripts/cli/base_interfaces_test.py +++ b/smoketest/scripts/cli/base_interfaces_test.py @@ -26,17 +26,24 @@ from base_vyostest_shim import VyOSUnitTestSHIM from vyos.configsession import ConfigSession from vyos.configsession import ConfigSessionError +from vyos.defaults import directories from vyos.ifconfig import Interface from vyos.ifconfig import Section from vyos.utils.file import read_file from vyos.utils.dict import dict_search from vyos.utils.process import process_named_running from vyos.utils.network import get_interface_config +from vyos.utils.network import get_interface_vrf from vyos.utils.process import cmd -from vyos.validate import is_intf_addr_assigned -from vyos.validate import is_ipv6_link_local +from vyos.utils.network import is_intf_addr_assigned +from vyos.utils.network import is_ipv6_link_local from vyos.xml_ref import cli_defined +dhclient_base_dir = directories['isc_dhclient_dir'] +dhclient_process_name = 'dhclient' +dhcp6c_base_dir = directories['dhcp6_client_dir'] +dhcp6c_process_name = 'dhcp6c' + def is_mirrored_to(interface, mirror_if, qdisc): """ Ask TC if we are mirroring traffic to a discrete interface. @@ -66,6 +73,7 @@ class BasicInterfaceTest: _test_ipv6_pd = False _test_ipv6_dhcpc6 = False _test_mirror = False + _test_vrf = False _base_path = [] _options = {} @@ -93,6 +101,7 @@ class BasicInterfaceTest: cls._test_ipv6_dhcpc6 = cli_defined(cls._base_path, 'dhcpv6-options') cls._test_ipv6_pd = cli_defined(cls._base_path + ['dhcpv6-options'], 'pd') cls._test_mtu = cli_defined(cls._base_path, 'mtu') + cls._test_vrf = cli_defined(cls._base_path, 'vrf') # Setup mirror interfaces for SPAN (Switch Port Analyzer) for span in cls._mirror_interfaces: @@ -137,8 +146,6 @@ class BasicInterfaceTest: for option in self._options.get(interface, []): self.cli_set(self._base_path + [interface] + option.split()) - self.cli_set(self._base_path + [interface, 'disable']) - # Also enable DHCP (ISC DHCP always places interface in admin up # state so we check that we do not start DHCP client. # https://vyos.dev/T2767 @@ -151,6 +158,99 @@ class BasicInterfaceTest: flags = read_file(f'/sys/class/net/{interface}/flags') self.assertEqual(int(flags, 16) & 1, 0) + def test_dhcp_client_options(self): + if not self._test_dhcp or not self._test_vrf: + self.skipTest('not supported') + + distance = '100' + + for interface in self._interfaces: + for option in self._options.get(interface, []): + self.cli_set(self._base_path + [interface] + option.split()) + + self.cli_set(self._base_path + [interface, 'address', 'dhcp']) + self.cli_set(self._base_path + [interface, 'dhcp-options', 'default-route-distance', distance]) + + self.cli_commit() + + for interface in self._interfaces: + # Check if dhclient process runs + dhclient_pid = process_named_running(dhclient_process_name, cmdline=interface) + self.assertTrue(dhclient_pid) + + dhclient_config = read_file(f'{dhclient_base_dir}/dhclient_{interface}.conf') + self.assertIn('request subnet-mask, broadcast-address, routers, domain-name-servers', dhclient_config) + self.assertIn('require subnet-mask;', dhclient_config) + + # and the commandline has the appropriate options + cmdline = read_file(f'/proc/{dhclient_pid}/cmdline') + self.assertIn(f'-e\x00IF_METRIC={distance}', cmdline) + + def test_dhcp_vrf(self): + if not self._test_dhcp or not self._test_vrf: + self.skipTest('not supported') + + vrf_name = 'purple4' + self.cli_set(['vrf', 'name', vrf_name, 'table', '65000']) + + for interface in self._interfaces: + for option in self._options.get(interface, []): + self.cli_set(self._base_path + [interface] + option.split()) + + self.cli_set(self._base_path + [interface, 'address', 'dhcp']) + self.cli_set(self._base_path + [interface, 'vrf', vrf_name]) + + self.cli_commit() + + # Validate interface state + for interface in self._interfaces: + tmp = get_interface_vrf(interface) + self.assertEqual(tmp, vrf_name) + + # Check if dhclient process runs + dhclient_pid = process_named_running(dhclient_process_name, cmdline=interface) + self.assertTrue(dhclient_pid) + # .. inside the appropriate VRF instance + vrf_pids = cmd(f'ip vrf pids {vrf_name}') + self.assertIn(str(dhclient_pid), vrf_pids) + # and the commandline has the appropriate options + cmdline = read_file(f'/proc/{dhclient_pid}/cmdline') + self.assertIn('-e\x00IF_METRIC=210', cmdline) # 210 is the default value + + self.cli_delete(['vrf', 'name', vrf_name]) + + def test_dhcpv6_vrf(self): + if not self._test_ipv6_dhcpc6 or not self._test_vrf: + self.skipTest('not supported') + + vrf_name = 'purple6' + self.cli_set(['vrf', 'name', vrf_name, 'table', '65001']) + + # When interface is configured as admin down, it must be admin down + # even when dhcpc starts on the given interface + for interface in self._interfaces: + for option in self._options.get(interface, []): + self.cli_set(self._base_path + [interface] + option.split()) + + self.cli_set(self._base_path + [interface, 'address', 'dhcpv6']) + self.cli_set(self._base_path + [interface, 'vrf', vrf_name]) + + self.cli_commit() + + # Validate interface state + for interface in self._interfaces: + tmp = get_interface_vrf(interface) + self.assertEqual(tmp, vrf_name) + + # Check if dhclient process runs + tmp = process_named_running(dhcp6c_process_name, cmdline=interface) + self.assertTrue(tmp) + # .. inside the appropriate VRF instance + vrf_pids = cmd(f'ip vrf pids {vrf_name}') + self.assertIn(str(tmp), vrf_pids) + + self.cli_delete(['vrf', 'name', vrf_name]) + def test_span_mirror(self): if not self._mirror_interfaces: self.skipTest('not supported') @@ -743,6 +843,7 @@ class BasicInterfaceTest: mss = '1400' dad_transmits = '10' + accept_dad = '0' for interface in self._interfaces: path = self._base_path + [interface] @@ -753,6 +854,9 @@ class BasicInterfaceTest: if cli_defined(self._base_path + ['ipv6'], 'adjust-mss'): self.cli_set(path + ['ipv6', 'adjust-mss', mss]) + if cli_defined(self._base_path + ['ipv6'], 'accept-dad'): + self.cli_set(path + ['ipv6', 'accept-dad', accept_dad]) + if cli_defined(self._base_path + ['ipv6'], 'dup-addr-detect-transmits'): self.cli_set(path + ['ipv6', 'dup-addr-detect-transmits', dad_transmits]) @@ -770,6 +874,10 @@ class BasicInterfaceTest: if line.startswith(base_options): self.assertIn(f'tcp option maxseg size set {mss}', line) + if cli_defined(self._base_path + ['ipv6'], 'accept-dad'): + tmp = read_file(f'{proc_base}/accept_dad') + self.assertEqual(accept_dad, tmp) + if cli_defined(self._base_path + ['ipv6'], 'dup-addr-detect-transmits'): tmp = read_file(f'{proc_base}/dad_transmits') self.assertEqual(dad_transmits, tmp) @@ -799,13 +907,10 @@ class BasicInterfaceTest: self.cli_commit() - dhcp6c_options = read_file(f'/run/dhcp6c/dhcp6c.{interface}.options') - self.assertIn(f'-n', dhcp6c_options) - duid_base = 10 for interface in self._interfaces: duid = '00:01:00:01:27:71:db:f0:00:50:00:00:00:{}'.format(duid_base) - dhcpc6_config = read_file(f'/run/dhcp6c/dhcp6c.{interface}.conf') + dhcpc6_config = read_file(f'{dhcp6c_base_dir}/dhcp6c.{interface}.conf') self.assertIn(f'interface {interface} ' + '{', dhcpc6_config) self.assertIn(f' request domain-name-servers;', dhcpc6_config) self.assertIn(f' request domain-name;', dhcpc6_config) @@ -816,8 +921,12 @@ class BasicInterfaceTest: self.assertIn('};', dhcpc6_config) duid_base += 1 - # Check for running process - self.assertTrue(process_named_running('dhcp6c')) + # Better ask the process about it's commandline in the future + pid = process_named_running(dhcp6c_process_name, cmdline=interface) + self.assertTrue(pid) + + dhcp6c_options = read_file(f'/proc/{pid}/cmdline') + self.assertIn('-n', dhcp6c_options) def test_dhcpv6pd_auto_sla_id(self): if not self._test_ipv6_pd: @@ -853,7 +962,7 @@ class BasicInterfaceTest: self.cli_commit() for interface in self._interfaces: - dhcpc6_config = read_file(f'/run/dhcp6c/dhcp6c.{interface}.conf') + dhcpc6_config = read_file(f'{dhcp6c_base_dir}/dhcp6c.{interface}.conf') # verify DHCPv6 prefix delegation self.assertIn(f'prefix ::/{prefix_len} infinity;', dhcpc6_config) @@ -871,8 +980,8 @@ class BasicInterfaceTest: # increment interface address address = str(int(address) + 1) - # Check for running process - self.assertTrue(process_named_running('dhcp6c')) + # Check for running process + self.assertTrue(process_named_running(dhcp6c_process_name, cmdline=interface)) for delegatee in delegatees: # we can already cleanup the test delegatee interface here @@ -921,7 +1030,7 @@ class BasicInterfaceTest: for interface in self._interfaces: address = '1' sla_id = '1' - dhcpc6_config = read_file(f'/run/dhcp6c/dhcp6c.{interface}.conf') + dhcpc6_config = read_file(f'{dhcp6c_base_dir}/dhcp6c.{interface}.conf') # verify DHCPv6 prefix delegation self.assertIn(f'prefix ::/{prefix_len} infinity;', dhcpc6_config) @@ -938,7 +1047,7 @@ class BasicInterfaceTest: address = str(int(address) + 1) # Check for running process - self.assertTrue(process_named_running('dhcp6c', interface)) + self.assertTrue(process_named_running(dhcp6c_process_name, cmdline=interface)) for delegatee in delegatees: # we can already cleanup the test delegatee interface here |