No OneTemporary
Actions

Size

23 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/smoketest/scripts/cli/base_interfaces_test.py b/smoketest/scripts/cli/base_interfaces_test.py
	index fbedca3f3..8b25d85e6 100644
	--- a/smoketest/scripts/cli/base_interfaces_test.py
	+++ b/smoketest/scripts/cli/base_interfaces_test.py
	@@ -1,515 +1,516 @@
	# Copyright (C) 2019-2021 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 <http://www.gnu.org/licenses/>.

	import os
	import unittest
	import json

	from binascii import hexlify

	-from netifaces import ifaddresses
	from netifaces import AF_INET
	from netifaces import AF_INET6
	+from netifaces import ifaddresses
	+from netifaces import interfaces

	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
	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

	class BasicInterfaceTest:
	class BaseTest(unittest.TestCase):
	_test_ip = False
	_test_mtu = False
	_test_vlan = False
	_test_qinq = False
	_test_ipv6 = False
	_test_ipv6_pd = False
	_test_mirror = False
	_base_path = []

	_options = {}
	_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())

	# 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

	# Ethernet is handled in its derived class
	if 'ethernet' not in self._base_path:
	for intf in self._interfaces:
	- self.assertFalse(os.path.isdir('/sys/class/net/{intf}'))
	+ self.assertTrue(intf not in interfaces())

	def test_span_mirror(self):
	if not self._mirror_interfaces:
	self.skipTest('not enabled')

	# 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_disable(self):
	# Check if description can be added to interface and
	# can be read back
	for intf in self._interfaces:
	self.session.set(self._base_path + [intf, 'disable'])
	for option in self._options.get(intf, []):
	self.session.set(self._base_path + [intf] + option.split())

	self.session.commit()

	# Validate interface description
	for intf in self._interfaces:
	self.assertEqual(Interface(intf).get_admin_state(), 'down')

	def test_interface_description(self):
	# Check if description can be added to interface and
	# can be read back
	for intf in self._interfaces:
	test_string=f'Description-Test-{intf}'
	self.session.set(self._base_path + [intf, 'description', test_string])
	for option in self._options.get(intf, []):
	self.session.set(self._base_path + [intf] + option.split())

	self.session.commit()

	# Validate interface description
	for intf in self._interfaces:
	test_string=f'Description-Test-{intf}'
	tmp = read_file(f'/sys/class/net/{intf}/ifalias')
	self.assertEqual(tmp, test_string)
	self.assertEqual(Interface(intf).get_alias(), test_string)
	self.session.delete(self._base_path + [intf, 'description'])

	self.session.commit()

	# Validate remove interface description "empty"
	for intf in self._interfaces:
	tmp = read_file(f'/sys/class/net/{intf}/ifalias')
	self.assertEqual(tmp, str())
	self.assertEqual(Interface(intf).get_alias(), str())

	def test_add_single_ip_address(self):
	addr = '192.0.2.0/31'
	for intf in self._interfaces:
	self.session.set(self._base_path + [intf, 'address', addr])
	for option in self._options.get(intf, []):
	self.session.set(self._base_path + [intf] + option.split())

	self.session.commit()

	for intf in self._interfaces:
	self.assertTrue(is_intf_addr_assigned(intf, addr))
	self.assertEqual(Interface(intf).get_admin_state(), 'up')

	def test_add_multiple_ip_addresses(self):
	# Add address
	for intf in self._interfaces:
	for addr in self._test_addr:
	self.session.set(self._base_path + [intf, 'address', addr])
	for option in self._options.get(intf, []):
	self.session.set(self._base_path + [intf] + option.split())

	self.session.commit()

	# Validate address
	for intf in self._interfaces:
	for af in AF_INET, AF_INET6:
	for addr in ifaddresses(intf)[af]:
	# checking link local addresses makes no sense
	if is_ipv6_link_local(addr['addr']):
	continue

	self.assertTrue(is_intf_addr_assigned(intf, addr['addr']))

	def test_ipv6_link_local_address(self):
	# Common function for IPv6 link-local address assignemnts
	if not self._test_ipv6:
	self.skipTest('not enabled')

	for interface in self._interfaces:
	base = self._base_path + [interface]
	for option in self._options.get(interface, []):
	self.session.set(base + option.split())

	# after commit we must have an IPv6 link-local address
	self.session.commit()

	for interface in self._interfaces:
	for addr in ifaddresses(interface)[AF_INET6]:
	self.assertTrue(is_ipv6_link_local(addr['addr']))

	# disable IPv6 link-local address assignment
	for interface in self._interfaces:
	base = self._base_path + [interface]
	self.session.set(base + ['ipv6', 'address', 'no-default-link-local'])

	# after commit we must have no IPv6 link-local address
	self.session.commit()

	for interface in self._interfaces:
	self.assertTrue(AF_INET6 not in ifaddresses(interface))

	def test_interface_mtu(self):
	if not self._test_mtu:
	self.skipTest('not enabled')

	for intf in self._interfaces:
	base = self._base_path + [intf]
	self.session.set(base + ['mtu', self._mtu])
	for option in self._options.get(intf, []):
	self.session.set(base + option.split())

	# commit interface changes
	self.session.commit()

	# verify changed MTU
	for intf in self._interfaces:
	tmp = read_file(f'/sys/class/net/{intf}/mtu')
	self.assertEqual(tmp, self._mtu)

	def test_mtu_1200_no_ipv6_interface(self):
	# Testcase if MTU can be changed to 1200 on non IPv6
	# enabled interfaces
	if not self._test_mtu:
	self.skipTest('not enabled')

	old_mtu = self._mtu
	self._mtu = '1200'

	for intf in self._interfaces:
	base = self._base_path + [intf]
	self.session.set(base + ['mtu', self._mtu])
	self.session.set(base + ['ipv6', 'address', 'no-default-link-local'])

	for option in self._options.get(intf, []):
	self.session.set(base + option.split())

	# commit interface changes
	self.session.commit()

	# verify changed MTU
	for intf in self._interfaces:
	tmp = read_file(f'/sys/class/net/{intf}/mtu')
	self.assertEqual(tmp, self._mtu)

	self._mtu = old_mtu

	def test_8021q_vlan_interfaces(self):
	if not self._test_vlan:
	self.skipTest('not enabled')

	for interface in self._interfaces:
	base = self._base_path + [interface]
	for option in self._options.get(interface, []):
	self.session.set(base + option.split())

	for vlan in self._vlan_range:
	base = self._base_path + [interface, 'vif', vlan]
	self.session.set(base + ['mtu', self._mtu])
	for address in self._test_addr:
	self.session.set(base + ['address', address])

	self.session.commit()

	for intf in self._interfaces:
	for vlan in self._vlan_range:
	vif = f'{intf}.{vlan}'
	for address in self._test_addr:
	self.assertTrue(is_intf_addr_assigned(vif, address))

	tmp = read_file(f'/sys/class/net/{vif}/mtu')
	self.assertEqual(tmp, self._mtu)
	self.assertEqual(Interface(vif).get_admin_state(), 'up')


	def test_8021ad_qinq_vlan_interfaces(self):
	if not self._test_qinq:
	self.skipTest('not enabled')

	for interface in self._interfaces:
	base = self._base_path + [interface]
	for option in self._options.get(interface, []):
	self.session.set(base + option.split())

	for vif_s in self._qinq_range:
	for vif_c in self._vlan_range:
	base = self._base_path + [interface, 'vif-s', vif_s, 'vif-c', vif_c]
	self.session.set(base + ['mtu', self._mtu])
	for address in self._test_addr:
	self.session.set(base + ['address', address])

	self.session.commit()

	for interface in self._interfaces:
	for vif_s in self._qinq_range:
	tmp = json.loads(cmd(f'ip -d -j link show dev {interface}.{vif_s}'))[0]
	self.assertEqual(dict_search('linkinfo.info_data.protocol', tmp), '802.1ad')

	for vif_c in self._vlan_range:
	vif = f'{interface}.{vif_s}.{vif_c}'
	for address in self._test_addr:
	self.assertTrue(is_intf_addr_assigned(vif, address))

	tmp = read_file(f'/sys/class/net/{vif}/mtu')
	self.assertEqual(tmp, self._mtu)

	def test_interface_ip_options(self):
	if not self._test_ip:
	self.skipTest('not enabled')

	for interface in self._interfaces:
	arp_tmo = '300'
	path = self._base_path + [interface]
	for option in self._options.get(interface, []):
	self.session.set(path + option.split())

	# Options
	self.session.set(path + ['ip', 'arp-cache-timeout', arp_tmo])
	self.session.set(path + ['ip', 'disable-arp-filter'])
	self.session.set(path + ['ip', 'disable-forwarding'])
	self.session.set(path + ['ip', 'enable-arp-accept'])
	self.session.set(path + ['ip', 'enable-arp-announce'])
	self.session.set(path + ['ip', 'enable-arp-ignore'])
	self.session.set(path + ['ip', 'enable-proxy-arp'])
	self.session.set(path + ['ip', 'proxy-arp-pvlan'])
	self.session.set(path + ['ip', 'source-validation', 'loose'])

	self.session.commit()

	for interface in self._interfaces:
	tmp = read_file(f'/proc/sys/net/ipv4/neigh/{interface}/base_reachable_time_ms')
	self.assertEqual(tmp, str((int(arp_tmo) * 1000))) # tmo value is in milli seconds

	tmp = read_file(f'/proc/sys/net/ipv4/conf/{interface}/arp_filter')
	self.assertEqual('0', tmp)

	tmp = read_file(f'/proc/sys/net/ipv4/conf/{interface}/arp_accept')
	self.assertEqual('1', tmp)

	tmp = read_file(f'/proc/sys/net/ipv4/conf/{interface}/arp_announce')
	self.assertEqual('1', tmp)

	tmp = read_file(f'/proc/sys/net/ipv4/conf/{interface}/arp_ignore')
	self.assertEqual('1', tmp)

	tmp = read_file(f'/proc/sys/net/ipv4/conf/{interface}/forwarding')
	self.assertEqual('0', tmp)

	tmp = read_file(f'/proc/sys/net/ipv4/conf/{interface}/proxy_arp')
	self.assertEqual('1', tmp)

	tmp = read_file(f'/proc/sys/net/ipv4/conf/{interface}/proxy_arp_pvlan')
	self.assertEqual('1', tmp)

	tmp = read_file(f'/proc/sys/net/ipv4/conf/{interface}/rp_filter')
	self.assertEqual('2', tmp)

	def test_interface_ipv6_options(self):
	if not self._test_ipv6:
	self.skipTest('not enabled')

	for interface in self._interfaces:
	dad_transmits = '10'
	path = self._base_path + [interface]
	for option in self._options.get(interface, []):
	self.session.set(path + option.split())

	# Options
	self.session.set(path + ['ipv6', 'disable-forwarding'])
	self.session.set(path + ['ipv6', 'dup-addr-detect-transmits', dad_transmits])

	self.session.commit()

	for interface in self._interfaces:
	tmp = read_file(f'/proc/sys/net/ipv6/conf/{interface}/forwarding')
	self.assertEqual('0', tmp)

	tmp = read_file(f'/proc/sys/net/ipv6/conf/{interface}/dad_transmits')
	self.assertEqual(dad_transmits, tmp)


	def test_dhcpv6pd_auto_sla_id(self):
	if not self._test_ipv6_pd:
	self.skipTest('not enabled')

	prefix_len = '56'
	sla_len = str(64 - int(prefix_len))

	delegatees = ['dum2340', 'dum2341', 'dum2342', 'dum2343', 'dum2344']

	for interface in self._interfaces:
	path = self._base_path + [interface]
	for option in self._options.get(interface, []):
	self.session.set(path + option.split())

	address = '1'
	# prefix delegation stuff
	pd_base = path + ['dhcpv6-options', 'pd', '0']
	self.session.set(pd_base + ['length', prefix_len])

	for delegatee in delegatees:
	section = Section.section(delegatee)
	self.session.set(['interfaces', section, delegatee])
	self.session.set(pd_base + ['interface', delegatee, 'address', address])
	# increment interface address
	address = str(int(address) + 1)

	self.session.commit()

	for interface in self._interfaces:
	dhcpc6_config = read_file(f'/run/dhcp6c/dhcp6c.{interface}.conf')

	# verify DHCPv6 prefix delegation
	self.assertIn(f'prefix ::/{prefix_len} infinity;', dhcpc6_config)

	address = '1'
	sla_id = '0'
	for delegatee in delegatees:
	self.assertIn(f'prefix-interface {delegatee}' + r' {', dhcpc6_config)
	self.assertIn(f'ifid {address};', dhcpc6_config)
	self.assertIn(f'sla-id {sla_id};', dhcpc6_config)
	self.assertIn(f'sla-len {sla_len};', dhcpc6_config)

	# increment sla-id
	sla_id = str(int(sla_id) + 1)
	# increment interface address
	address = str(int(address) + 1)

	# Check for running process
	self.assertTrue(process_named_running('dhcp6c'))

	for delegatee in delegatees:
	# we can already cleanup the test delegatee interface here
	# as until commit() is called, nothing happens
	section = Section.section(delegatee)
	self.session.delete(['interfaces', section, delegatee])

	def test_dhcpv6pd_manual_sla_id(self):
	if not self._test_ipv6_pd:
	self.skipTest('not enabled')

	prefix_len = '56'
	sla_len = str(64 - int(prefix_len))

	delegatees = ['dum3340', 'dum3341', 'dum3342', 'dum3343', 'dum3344']

	for interface in self._interfaces:
	path = self._base_path + [interface]
	for option in self._options.get(interface, []):
	self.session.set(path + option.split())

	# prefix delegation stuff
	address = '1'
	sla_id = '1'
	pd_base = path + ['dhcpv6-options', 'pd', '0']
	self.session.set(pd_base + ['length', prefix_len])

	for delegatee in delegatees:
	section = Section.section(delegatee)
	self.session.set(['interfaces', section, delegatee])
	self.session.set(pd_base + ['interface', delegatee, 'address', address])
	self.session.set(pd_base + ['interface', delegatee, 'sla-id', sla_id])

	# increment interface address
	address = str(int(address) + 1)
	sla_id = str(int(sla_id) + 1)

	self.session.commit()

	# Verify dhcpc6 client configuration
	for interface in self._interfaces:
	address = '1'
	sla_id = '1'
	dhcpc6_config = read_file(f'/run/dhcp6c/dhcp6c.{interface}.conf')

	# verify DHCPv6 prefix delegation
	self.assertIn(f'prefix ::/{prefix_len} infinity;', dhcpc6_config)

	for delegatee in delegatees:
	self.assertIn(f'prefix-interface {delegatee}' + r' {', dhcpc6_config)
	self.assertIn(f'ifid {address};', dhcpc6_config)
	self.assertIn(f'sla-id {sla_id};', dhcpc6_config)
	self.assertIn(f'sla-len {sla_len};', dhcpc6_config)

	# increment sla-id
	sla_id = str(int(sla_id) + 1)
	# increment interface address
	address = str(int(address) + 1)

	# Check for running process
	self.assertTrue(process_named_running('dhcp6c'))

	for delegatee in delegatees:
	# we can already cleanup the test delegatee interface here
	# as until commit() is called, nothing happens
	section = Section.section(delegatee)
	self.session.delete(['interfaces', section, delegatee])
	diff --git a/smoketest/scripts/cli/test_interfaces_loopback.py b/smoketest/scripts/cli/test_interfaces_loopback.py
	index 79225a1bd..e3538fc59 100755
	--- a/smoketest/scripts/cli/test_interfaces_loopback.py
	+++ b/smoketest/scripts/cli/test_interfaces_loopback.py
	@@ -1,41 +1,52 @@
	#!/usr/bin/env python3
	#
	# Copyright (C) 2020 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 <http://www.gnu.org/licenses/>.

	import unittest

	from base_interfaces_test import BasicInterfaceTest
	+from netifaces import interfaces
	+
	from vyos.validate import is_intf_addr_assigned

	class LoopbackInterfaceTest(BasicInterfaceTest.BaseTest):
	def setUp(self):
	super().setUp()
	# these addresses are never allowed to be removed from the system
	self._loopback_addresses = ['127.0.0.1', '::1']
	self._base_path = ['interfaces', 'loopback']
	self._interfaces = ['lo']

	+ def tearDown(self):
	+ self.session.delete(self._base_path)
	+ self.session.commit()
	+ del self.session
	+
	+ # loopback interface must persist!
	+ for intf in self._interfaces:
	+ self.assertTrue(intf in interfaces())
	+
	def test_add_single_ip_address(self):
	super().test_add_single_ip_address()
	for addr in self._loopback_addresses:
	self.assertTrue(is_intf_addr_assigned('lo', addr))

	def test_add_multiple_ip_addresses(self):
	super().test_add_multiple_ip_addresses()
	for addr in self._loopback_addresses:
	self.assertTrue(is_intf_addr_assigned('lo', addr))

	if __name__ == '__main__':
	unittest.main(verbosity=2)

File Metadata

Mime Type: text/x-diff
Expires: Fri, Nov 6, 7:04 AM (14 h, 29 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 4652980
Default Alt Text: (23 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions