diff options
Diffstat (limited to 'smoketest')
| -rw-r--r-- | smoketest/scripts/cli/base_interfaces_test.py | 16 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_firewall.py | 38 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_interfaces_macsec.py | 35 |
3 files changed, 72 insertions, 17 deletions
diff --git a/smoketest/scripts/cli/base_interfaces_test.py b/smoketest/scripts/cli/base_interfaces_test.py index 277f14067..6862207e3 100644 --- a/smoketest/scripts/cli/base_interfaces_test.py +++ b/smoketest/scripts/cli/base_interfaces_test.py @@ -158,14 +158,22 @@ class BasicInterfaceTest: if not self._test_dhcp or not self._test_vrf: self.skipTest('not supported') + client_id = 'VyOS-router' distance = '100' + hostname = 'vyos' + vendor_class_id = 'vyos-vendor' + user_class = 'vyos' 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', 'client-id', client_id]) self.cli_set(self._base_path + [interface, 'dhcp-options', 'default-route-distance', distance]) + self.cli_set(self._base_path + [interface, 'dhcp-options', 'host-name', hostname]) + self.cli_set(self._base_path + [interface, 'dhcp-options', 'vendor-class-id', vendor_class_id]) + self.cli_set(self._base_path + [interface, 'dhcp-options', 'user-class', user_class]) self.cli_commit() @@ -175,8 +183,12 @@ class BasicInterfaceTest: 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) + self.assertIn(f'request subnet-mask, broadcast-address, routers, domain-name-servers', dhclient_config) + self.assertIn(f'require subnet-mask;', dhclient_config) + self.assertIn(f'send host-name "{hostname}";', dhclient_config) + self.assertIn(f'send dhcp-client-identifier "{client_id}";', dhclient_config) + self.assertIn(f'send vendor-class-identifier "{vendor_class_id}";', dhclient_config) + self.assertIn(f'send user-class "{user_class}";', dhclient_config) # and the commandline has the appropriate options cmdline = read_file(f'/proc/{dhclient_pid}/cmdline') diff --git a/smoketest/scripts/cli/test_firewall.py b/smoketest/scripts/cli/test_firewall.py index e6c928ad7..18940c04d 100755 --- a/smoketest/scripts/cli/test_firewall.py +++ b/smoketest/scripts/cli/test_firewall.py @@ -653,5 +653,43 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase): break self.assertTrue(matched) + def test_flow_offload(self): + self.cli_set(['firewall', 'flowtable', 'smoketest', 'interface', 'eth0']) + self.cli_set(['firewall', 'flowtable', 'smoketest', 'offload', 'hardware']) + + # QEMU virtual NIC does not support hw-tc-offload + with self.assertRaises(ConfigSessionError): + self.cli_commit() + + self.cli_set(['firewall', 'flowtable', 'smoketest', 'offload', 'software']) + + self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'action', 'offload']) + self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'offload-target', 'smoketest']) + self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'protocol', 'tcp_udp']) + self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'state', 'established']) + self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'state', 'related']) + + self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '1', 'action', 'offload']) + self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '1', 'offload-target', 'smoketest']) + self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '1', 'protocol', 'tcp_udp']) + self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '1', 'state', 'established']) + self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '1', 'state', 'related']) + + self.cli_commit() + + nftables_search = [ + ['flowtable VYOS_FLOWTABLE_smoketest'], + ['hook ingress priority filter'], + ['devices = { eth0 }'], + ['ct state { established, related }', 'meta l4proto { tcp, udp }', 'flow add @VYOS_FLOWTABLE_smoketest'], + ] + + self.verify_nftables(nftables_search, 'ip vyos_filter') + self.verify_nftables(nftables_search, 'ip6 vyos_filter') + + # Check conntrack + #self.verify_nftables_chain([['accept']], 'ip vyos_conntrack', 'FW_CONNTRACK') + #self.verify_nftables_chain([['accept']], 'ip6 vyos_conntrack', 'FW_CONNTRACK') + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/smoketest/scripts/cli/test_interfaces_macsec.py b/smoketest/scripts/cli/test_interfaces_macsec.py index ea0f00071..d8d564792 100755 --- a/smoketest/scripts/cli/test_interfaces_macsec.py +++ b/smoketest/scripts/cli/test_interfaces_macsec.py @@ -14,7 +14,6 @@ # 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 re import unittest @@ -23,9 +22,10 @@ from netifaces import interfaces from vyos.configsession import ConfigSessionError from vyos.ifconfig import Section -from vyos.utils.process import cmd from vyos.utils.file import read_file from vyos.utils.network import get_interface_config +from vyos.utils.network import interface_exists +from vyos.utils.process import cmd from vyos.utils.process import process_named_running PROCESS_NAME = 'wpa_supplicant' @@ -35,10 +35,6 @@ def get_config_value(interface, key): tmp = re.findall(r'\n?{}=(.*)'.format(key), tmp) return tmp[0] -def get_cipher(interface): - tmp = get_interface_config(interface) - return tmp['linkinfo']['info_data']['cipher_suite'].lower() - class MACsecInterfaceTest(BasicInterfaceTest.TestCase): @classmethod def setUpClass(cls): @@ -117,6 +113,10 @@ class MACsecInterfaceTest(BasicInterfaceTest.TestCase): tmp = read_file(f'/sys/class/net/{interface}/mtu') self.assertEqual(tmp, '1460') + # Encryption enabled? + tmp = get_interface_config(interface) + self.assertTrue(tmp['linkinfo']['info_data']['encrypt']) + # Check for running process self.assertTrue(process_named_running(PROCESS_NAME)) @@ -138,10 +138,11 @@ class MACsecInterfaceTest(BasicInterfaceTest.TestCase): # final commit and verify self.cli_commit() - self.assertIn(interface, interfaces()) + self.assertTrue(interface_exists(interface)) # Verify proper cipher suite (T4537) - self.assertEqual(cipher, get_cipher(interface)) + tmp = get_interface_config(interface) + self.assertEqual(cipher, tmp['linkinfo']['info_data']['cipher_suite'].lower()) def test_macsec_gcm_aes_256(self): src_interface = 'eth0' @@ -161,10 +162,11 @@ class MACsecInterfaceTest(BasicInterfaceTest.TestCase): # final commit and verify self.cli_commit() - self.assertIn(interface, interfaces()) + self.assertTrue(interface_exists(interface)) # Verify proper cipher suite (T4537) - self.assertEqual(cipher, get_cipher(interface)) + tmp = get_interface_config(interface) + self.assertEqual(cipher, tmp['linkinfo']['info_data']['cipher_suite'].lower()) def test_macsec_source_interface(self): # Ensure source-interface can bot be part of any other bond or bridge @@ -191,7 +193,7 @@ class MACsecInterfaceTest(BasicInterfaceTest.TestCase): # final commit and verify self.cli_commit() - self.assertIn(interface, interfaces()) + self.assertTrue(interface_exists(interface)) def test_macsec_static_keys(self): src_interface = 'eth0' @@ -205,7 +207,7 @@ class MACsecInterfaceTest(BasicInterfaceTest.TestCase): peer_mac = '00:11:22:33:44:55' self.cli_set(self._base_path + [interface]) - # Encrypt link + # Encrypt link self.cli_set(self._base_path + [interface, 'security', 'encrypt']) # check validate() - source interface is mandatory @@ -261,9 +263,12 @@ class MACsecInterfaceTest(BasicInterfaceTest.TestCase): # final commit and verify self.cli_commit() - self.assertIn(interface, interfaces()) - self.assertEqual(cipher2, get_cipher(interface)) - self.assertTrue(os.path.isdir(f'/sys/class/net/{interface}')) + + self.assertTrue(interface_exists(interface)) + tmp = get_interface_config(interface) + self.assertEqual(cipher2, tmp['linkinfo']['info_data']['cipher_suite'].lower()) + # Encryption enabled? + self.assertTrue(tmp['linkinfo']['info_data']['encrypt']) if __name__ == '__main__': unittest.main(verbosity=2) |