diff options
Diffstat (limited to 'smoketest/scripts/cli/base_interfaces_test.py')
| -rw-r--r-- | smoketest/scripts/cli/base_interfaces_test.py | 131 | 
1 files changed, 116 insertions, 15 deletions
diff --git a/smoketest/scripts/cli/base_interfaces_test.py b/smoketest/scripts/cli/base_interfaces_test.py index a3868fa70..cef61e42b 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') @@ -799,13 +899,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 +913,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 +954,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 +972,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 +1022,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 +1039,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  | 
