From 710dac553fac93d8a205c9bc7e6b116753ac0b34 Mon Sep 17 00:00:00 2001 From: Christian Breunig Date: Sun, 6 Aug 2023 19:05:34 +0200 Subject: smoketest: T5428: check for process running in designated VRF Start IPv4/IPv6 DHCP clients on an interface bound to a given VRF. Verify that the client process runs in the VRF context. --- smoketest/scripts/cli/base_interfaces_test.py | 105 +++++++++++++++++++++++++- src/tests/test_util.py | 28 ------- src/tests/test_utils.py | 28 +++++++ src/tests/test_utils_network.py | 50 ++++++++++++ src/tests/test_validate.py | 50 ------------ 5 files changed, 180 insertions(+), 81 deletions(-) delete mode 100644 src/tests/test_util.py create mode 100644 src/tests/test_utils.py create mode 100644 src/tests/test_utils_network.py delete mode 100644 src/tests/test_validate.py diff --git a/smoketest/scripts/cli/base_interfaces_test.py b/smoketest/scripts/cli/base_interfaces_test.py index d4652a5ad..4244410bb 100644 --- a/smoketest/scripts/cli/base_interfaces_test.py +++ b/smoketest/scripts/cli/base_interfaces_test.py @@ -33,12 +33,16 @@ 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.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): """ @@ -69,6 +73,7 @@ class BasicInterfaceTest: _test_ipv6_pd = False _test_ipv6_dhcpc6 = False _test_mirror = False + _test_vrf = False _base_path = [] _options = {} @@ -96,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: @@ -154,6 +160,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') @@ -817,7 +916,7 @@ class BasicInterfaceTest: duid_base += 1 # Better ask the process about it's commandline in the future - pid = process_named_running('dhcp6c', cmdline=interface) + pid = process_named_running(dhcp6c_process_name, cmdline=interface) self.assertTrue(pid) dhcp6c_options = read_file(f'/proc/{pid}/cmdline') @@ -876,7 +975,7 @@ class BasicInterfaceTest: address = str(int(address) + 1) # Check for running process - self.assertTrue(process_named_running('dhcp6c', cmdline=interface)) + self.assertTrue(process_named_running(dhcp6c_process_name, cmdline=interface)) for delegatee in delegatees: # we can already cleanup the test delegatee interface here @@ -942,7 +1041,7 @@ class BasicInterfaceTest: address = str(int(address) + 1) # Check for running process - self.assertTrue(process_named_running('dhcp6c', cmdline=interface)) + self.assertTrue(process_named_running(dhcp6c_process_name, cmdline=interface)) for delegatee in delegatees: # we can already cleanup the test delegatee interface here diff --git a/src/tests/test_util.py b/src/tests/test_util.py deleted file mode 100644 index 9ae329ced..000000000 --- a/src/tests/test_util.py +++ /dev/null @@ -1,28 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 2020-2023 VyOS maintainers and contributors -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 or later as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - -from unittest import TestCase -class TestVyOSUtils(TestCase): - def test_key_mangling(self): - from vyos.utils.dict import mangle_dict_keys - data = {"foo-bar": {"baz-quux": None}} - expected_data = {"foo_bar": {"baz_quux": None}} - new_data = mangle_dict_keys(data, '-', '_') - self.assertEqual(new_data, expected_data) - - def test_sysctl_read(self): - from vyos.utils.system import sysctl_read - self.assertEqual(sysctl_read('net.ipv4.conf.lo.forwarding'), '1') diff --git a/src/tests/test_utils.py b/src/tests/test_utils.py new file mode 100644 index 000000000..9ae329ced --- /dev/null +++ b/src/tests/test_utils.py @@ -0,0 +1,28 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2020-2023 VyOS maintainers and contributors +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 or later as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +from unittest import TestCase +class TestVyOSUtils(TestCase): + def test_key_mangling(self): + from vyos.utils.dict import mangle_dict_keys + data = {"foo-bar": {"baz-quux": None}} + expected_data = {"foo_bar": {"baz_quux": None}} + new_data = mangle_dict_keys(data, '-', '_') + self.assertEqual(new_data, expected_data) + + def test_sysctl_read(self): + from vyos.utils.system import sysctl_read + self.assertEqual(sysctl_read('net.ipv4.conf.lo.forwarding'), '1') diff --git a/src/tests/test_utils_network.py b/src/tests/test_utils_network.py new file mode 100644 index 000000000..5a6dc2586 --- /dev/null +++ b/src/tests/test_utils_network.py @@ -0,0 +1,50 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2020-2023 VyOS maintainers and contributors +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 or later as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import vyos.utils.network +from unittest import TestCase + +class TestVyOSUtilsNetwork(TestCase): + def setUp(self): + pass + + def test_is_addr_assigned(self): + self.assertTrue(vyos.utils.network.is_addr_assigned('127.0.0.1')) + self.assertTrue(vyos.utils.network.is_addr_assigned('::1')) + self.assertFalse(vyos.utils.network.is_addr_assigned('127.251.255.123')) + + def test_is_ipv6_link_local(self): + self.assertFalse(vyos.utils.network.is_ipv6_link_local('169.254.0.1')) + self.assertTrue(vyos.utils.network.is_ipv6_link_local('fe80::')) + self.assertTrue(vyos.utils.network.is_ipv6_link_local('fe80::affe:1')) + self.assertTrue(vyos.utils.network.is_ipv6_link_local('fe80::affe:1%eth0')) + self.assertFalse(vyos.utils.network.is_ipv6_link_local('2001:db8::')) + self.assertFalse(vyos.utils.network.is_ipv6_link_local('2001:db8::%eth0')) + self.assertFalse(vyos.utils.network.is_ipv6_link_local('VyOS')) + self.assertFalse(vyos.utils.network.is_ipv6_link_local('::1')) + self.assertFalse(vyos.utils.network.is_ipv6_link_local('::1%lo')) + + def test_is_ipv6_link_local(self): + self.assertTrue(vyos.utils.network.is_loopback_addr('127.0.0.1')) + self.assertTrue(vyos.utils.network.is_loopback_addr('127.0.1.1')) + self.assertTrue(vyos.utils.network.is_loopback_addr('127.1.1.1')) + self.assertTrue(vyos.utils.network.is_loopback_addr('::1')) + + self.assertFalse(vyos.utils.network.is_loopback_addr('::2')) + self.assertFalse(vyos.utils.network.is_loopback_addr('192.0.2.1')) + + + diff --git a/src/tests/test_validate.py b/src/tests/test_validate.py deleted file mode 100644 index 5a6dc2586..000000000 --- a/src/tests/test_validate.py +++ /dev/null @@ -1,50 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 2020-2023 VyOS maintainers and contributors -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 or later as -# published by the Free Software Foundation. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . - -import vyos.utils.network -from unittest import TestCase - -class TestVyOSUtilsNetwork(TestCase): - def setUp(self): - pass - - def test_is_addr_assigned(self): - self.assertTrue(vyos.utils.network.is_addr_assigned('127.0.0.1')) - self.assertTrue(vyos.utils.network.is_addr_assigned('::1')) - self.assertFalse(vyos.utils.network.is_addr_assigned('127.251.255.123')) - - def test_is_ipv6_link_local(self): - self.assertFalse(vyos.utils.network.is_ipv6_link_local('169.254.0.1')) - self.assertTrue(vyos.utils.network.is_ipv6_link_local('fe80::')) - self.assertTrue(vyos.utils.network.is_ipv6_link_local('fe80::affe:1')) - self.assertTrue(vyos.utils.network.is_ipv6_link_local('fe80::affe:1%eth0')) - self.assertFalse(vyos.utils.network.is_ipv6_link_local('2001:db8::')) - self.assertFalse(vyos.utils.network.is_ipv6_link_local('2001:db8::%eth0')) - self.assertFalse(vyos.utils.network.is_ipv6_link_local('VyOS')) - self.assertFalse(vyos.utils.network.is_ipv6_link_local('::1')) - self.assertFalse(vyos.utils.network.is_ipv6_link_local('::1%lo')) - - def test_is_ipv6_link_local(self): - self.assertTrue(vyos.utils.network.is_loopback_addr('127.0.0.1')) - self.assertTrue(vyos.utils.network.is_loopback_addr('127.0.1.1')) - self.assertTrue(vyos.utils.network.is_loopback_addr('127.1.1.1')) - self.assertTrue(vyos.utils.network.is_loopback_addr('::1')) - - self.assertFalse(vyos.utils.network.is_loopback_addr('::2')) - self.assertFalse(vyos.utils.network.is_loopback_addr('192.0.2.1')) - - - -- cgit v1.2.3