summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--python/vyos/validate.py33
-rw-r--r--smoketest/scripts/cli/base_interfaces_test.py56
-rwxr-xr-xsmoketest/scripts/cli/test_interfaces_bonding.py12
-rwxr-xr-xsmoketest/scripts/cli/test_interfaces_bridge.py12
-rwxr-xr-xsmoketest/scripts/cli/test_interfaces_dummy.py2
-rwxr-xr-xsmoketest/scripts/cli/test_interfaces_ethernet.py9
-rwxr-xr-xsmoketest/scripts/cli/test_interfaces_geneve.py8
-rwxr-xr-xsmoketest/scripts/cli/test_interfaces_l2tpv3.py4
-rwxr-xr-xsmoketest/scripts/cli/test_interfaces_pseudo_ethernet.py7
-rwxr-xr-xsmoketest/scripts/cli/test_interfaces_tunnel.py6
-rwxr-xr-xsmoketest/scripts/cli/test_interfaces_vxlan.py3
-rwxr-xr-xsmoketest/scripts/cli/test_interfaces_wireless.py3
-rwxr-xr-xsmoketest/scripts/cli/test_protocols_bgp.py3
13 files changed, 99 insertions, 59 deletions
diff --git a/python/vyos/validate.py b/python/vyos/validate.py
index acd6086ff..23e88b5ac 100644
--- a/python/vyos/validate.py
+++ b/python/vyos/validate.py
@@ -13,9 +13,6 @@
# You should have received a copy of the GNU Lesser General Public
# License along with this library. If not, see <http://www.gnu.org/licenses/>.
-import netifaces
-from vyos.util import cmd
-
# Important note when you are adding new validation functions:
#
# The Control class will analyse the signature of the function in this file
@@ -59,6 +56,9 @@ def _is_intf_addr_assigned(intf, address, netmask=''):
address 192.0.2.1/24.
"""
from vyos.template import is_ipv4
+ from netifaces import ifaddresses
+ from netifaces import AF_INET
+ from netifaces import AF_INET6
# check if the requested address type is configured at all
# {
@@ -67,13 +67,13 @@ def _is_intf_addr_assigned(intf, address, netmask=''):
# 10: [{'addr': 'fe80::a00:27ff:fed9:5b04%eth0', 'netmask': 'ffff:ffff:ffff:ffff::'}]
# }
try:
- ifaces = netifaces.ifaddresses(intf)
+ ifaces = ifaddresses(intf)
except ValueError as e:
print(e)
return False
# determine IP version (AF_INET or AF_INET6) depending on passed address
- addr_type = netifaces.AF_INET if is_ipv4(address) else netifaces.AF_INET6
+ addr_type = AF_INET if is_ipv4(address) else AF_INET6
# Check every IP address on this interface for a match
for ip in ifaces.get(addr_type,[]):
@@ -103,8 +103,8 @@ def is_addr_assigned(addr):
"""
Verify if the given IPv4/IPv6 address is assigned to any interface
"""
-
- for intf in netifaces.interfaces():
+ from netifaces import interfaces
+ for intf in interfaces():
tmp = is_intf_addr_assigned(intf, addr)
if tmp == True:
return True
@@ -129,28 +129,34 @@ def is_subnet_connected(subnet, primary=False):
"""
from ipaddress import ip_address
from ipaddress import ip_network
+
+ from netifaces import ifaddresses
+ from netifaces import interfaces
+ from netifaces import AF_INET
+ from netifaces import AF_INET6
+
from vyos.template import is_ipv6
# determine IP version (AF_INET or AF_INET6) depending on passed address
- addr_type = netifaces.AF_INET
+ addr_type = AF_INET
if is_ipv6(subnet):
- addr_type = netifaces.AF_INET6
+ addr_type = AF_INET6
- for interface in netifaces.interfaces():
+ for interface in interfaces():
# check if the requested address type is configured at all
- if addr_type not in netifaces.ifaddresses(interface).keys():
+ if addr_type not in ifaddresses(interface).keys():
continue
# An interface can have multiple addresses, but some software components
# only support the primary address :(
if primary:
- ip = netifaces.ifaddresses(interface)[addr_type][0]['addr']
+ ip = ifaddresses(interface)[addr_type][0]['addr']
if ip_address(ip) in ip_network(subnet):
return True
else:
# Check every assigned IP address if it is connected to the subnet
# in question
- for ip in netifaces.ifaddresses(interface)[addr_type]:
+ for ip in ifaddresses(interface)[addr_type]:
# remove interface extension (e.g. %eth0) that gets thrown on the end of _some_ addrs
addr = ip['addr'].split('%')[0]
if ip_address(addr) in ip_network(subnet):
@@ -190,6 +196,7 @@ def assert_mtu(mtu, ifname):
assert_number(mtu)
import json
+ from vyos.util import cmd
out = cmd(f'ip -j -d link show dev {ifname}')
# [{"ifindex":2,"ifname":"eth0","flags":["BROADCAST","MULTICAST","UP","LOWER_UP"],"mtu":1500,"qdisc":"pfifo_fast","operstate":"UP","linkmode":"DEFAULT","group":"default","txqlen":1000,"link_type":"ether","address":"08:00:27:d9:5b:04","broadcast":"ff:ff:ff:ff:ff:ff","promiscuity":0,"min_mtu":46,"max_mtu":16110,"inet6_addr_gen_mode":"none","num_tx_queues":1,"num_rx_queues":1,"gso_max_size":65536,"gso_max_segs":65535}]
parsed = json.loads(out)[0]
diff --git a/smoketest/scripts/cli/base_interfaces_test.py b/smoketest/scripts/cli/base_interfaces_test.py
index a784140f3..8ee5395d0 100644
--- a/smoketest/scripts/cli/base_interfaces_test.py
+++ b/smoketest/scripts/cli/base_interfaces_test.py
@@ -25,6 +25,7 @@ from netifaces import AF_INET6
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
@@ -32,6 +33,25 @@ 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
+
+
dhcp6c_config_file = '/run/dhcp6c/dhcp6c.{}.conf'
def get_dhcp6c_config_value(interface, key):
tmp = read_file(dhcp6c_config_file.format(interface))
@@ -56,25 +76,53 @@ class BasicInterfaceTest:
_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())
- self._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']
- self._test_mtu = False
- self._options = {}
+ # 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
+ def test_span_mirror(self):
+ if not self._mirror_interfaces:
+ return None
+
+ # 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_description(self):
# Check if description can be added to interface and
# can be read back
diff --git a/smoketest/scripts/cli/test_interfaces_bonding.py b/smoketest/scripts/cli/test_interfaces_bonding.py
index d38e11a63..a35682b7c 100755
--- a/smoketest/scripts/cli/test_interfaces_bonding.py
+++ b/smoketest/scripts/cli/test_interfaces_bonding.py
@@ -26,17 +26,15 @@ from vyos.util import read_file
class BondingInterfaceTest(BasicInterfaceTest.BaseTest):
def setUp(self):
- super().setUp()
-
- self._base_path = ['interfaces', 'bonding']
- self._interfaces = ['bond0']
self._test_mtu = True
self._test_vlan = True
self._test_qinq = True
self._test_ipv6 = True
- self._test_mirror = True
-
+ self._base_path = ['interfaces', 'bonding']
+ self._interfaces = ['bond0']
+ self._mirror_interfaces = ['dum21354']
self._members = []
+
# we need to filter out VLAN interfaces identified by a dot (.)
# in their name - just in case!
if 'TEST_ETH' in os.environ:
@@ -50,6 +48,8 @@ class BondingInterfaceTest(BasicInterfaceTest.BaseTest):
for member in self._members:
self._options['bond0'].append(f'member interface {member}')
+ super().setUp()
+
def test_add_single_ip_address(self):
super().test_add_single_ip_address()
diff --git a/smoketest/scripts/cli/test_interfaces_bridge.py b/smoketest/scripts/cli/test_interfaces_bridge.py
index 394d50025..7444701c1 100755
--- a/smoketest/scripts/cli/test_interfaces_bridge.py
+++ b/smoketest/scripts/cli/test_interfaces_bridge.py
@@ -28,17 +28,13 @@ from vyos.util import read_file
class BridgeInterfaceTest(BasicInterfaceTest.BaseTest):
def setUp(self):
- super().setUp()
-
self._test_ipv6 = True
self._test_vlan = True
self._test_qinq = True
- self._test_mirror = True
-
self._base_path = ['interfaces', 'bridge']
- self._interfaces = ['br0']
-
+ self._mirror_interfaces = ['dum21354']
self._members = []
+
# we need to filter out VLAN interfaces identified by a dot (.)
# in their name - just in case!
if 'TEST_ETH' in os.environ:
@@ -51,7 +47,9 @@ class BridgeInterfaceTest(BasicInterfaceTest.BaseTest):
self._options['br0'] = []
for member in self._members:
self._options['br0'].append(f'member interface {member}')
+ self._interfaces = list(self._options)
+ super().setUp()
def test_add_remove_bridge_member(self):
# Add member interfaces to bridge and set STP cost/priority
@@ -188,5 +186,5 @@ class BridgeInterfaceTest(BasicInterfaceTest.BaseTest):
self.session.delete(['interfaces', 'ethernet', member, 'vif', vif])
if __name__ == '__main__':
- unittest.main(verbosity=2)
+ unittest.main(verbosity=2, failfast=True)
diff --git a/smoketest/scripts/cli/test_interfaces_dummy.py b/smoketest/scripts/cli/test_interfaces_dummy.py
index 97f5344ac..c482a6f0b 100755
--- a/smoketest/scripts/cli/test_interfaces_dummy.py
+++ b/smoketest/scripts/cli/test_interfaces_dummy.py
@@ -20,9 +20,9 @@ from base_interfaces_test import BasicInterfaceTest
class DummyInterfaceTest(BasicInterfaceTest.BaseTest):
def setUp(self):
- super().setUp()
self._base_path = ['interfaces', 'dummy']
self._interfaces = ['dum0', 'dum1', 'dum2']
+ super().setUp()
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_interfaces_ethernet.py b/smoketest/scripts/cli/test_interfaces_ethernet.py
index 2d0a4827d..3c4796283 100755
--- a/smoketest/scripts/cli/test_interfaces_ethernet.py
+++ b/smoketest/scripts/cli/test_interfaces_ethernet.py
@@ -35,16 +35,13 @@ def get_wpa_supplicant_value(interface, key):
class EthernetInterfaceTest(BasicInterfaceTest.BaseTest):
def setUp(self):
- super().setUp()
-
- self._base_path = ['interfaces', 'ethernet']
self._test_ip = True
self._test_mtu = True
self._test_vlan = True
self._test_qinq = True
self._test_ipv6 = True
- self._test_mirror = True
- self._interfaces = []
+ self._base_path = ['interfaces', 'ethernet']
+ self._mirror_interfaces = ['dum21354']
# we need to filter out VLAN interfaces identified by a dot (.)
# in their name - just in case!
@@ -66,6 +63,8 @@ class EthernetInterfaceTest(BasicInterfaceTest.BaseTest):
mac = read_file(f'/sys/class/net/{interface}/address')
self._macs[interface] = mac
+ super().setUp()
+
def tearDown(self):
for interface in self._interfaces:
diff --git a/smoketest/scripts/cli/test_interfaces_geneve.py b/smoketest/scripts/cli/test_interfaces_geneve.py
index 7e0389a63..98f55210f 100755
--- a/smoketest/scripts/cli/test_interfaces_geneve.py
+++ b/smoketest/scripts/cli/test_interfaces_geneve.py
@@ -14,24 +14,20 @@
# 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
-from vyos.configsession import ConfigSession, ConfigSessionError
+from vyos.configsession import ConfigSession
from base_interfaces_test import BasicInterfaceTest
-
class GeneveInterfaceTest(BasicInterfaceTest.BaseTest):
def setUp(self):
- super().setUp()
-
self._base_path = ['interfaces', 'geneve']
self._options = {
'gnv0': ['vni 10', 'remote 127.0.1.1'],
'gnv1': ['vni 20', 'remote 127.0.1.2'],
}
self._interfaces = list(self._options)
-
+ super().setUp()
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_interfaces_l2tpv3.py b/smoketest/scripts/cli/test_interfaces_l2tpv3.py
index be9565d00..c756bfdd5 100755
--- a/smoketest/scripts/cli/test_interfaces_l2tpv3.py
+++ b/smoketest/scripts/cli/test_interfaces_l2tpv3.py
@@ -15,7 +15,6 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import json
-import jmespath
import unittest
from base_interfaces_test import BasicInterfaceTest
@@ -23,8 +22,6 @@ from vyos.util import cmd
class GeneveInterfaceTest(BasicInterfaceTest.BaseTest):
def setUp(self):
- super().setUp()
-
self._base_path = ['interfaces', 'l2tpv3']
self._options = {
'l2tpeth10': ['local-ip 127.0.0.1', 'remote-ip 127.10.10.10',
@@ -37,6 +34,7 @@ class GeneveInterfaceTest(BasicInterfaceTest.BaseTest):
'source-port 2020', 'destination-port 20202'],
}
self._interfaces = list(self._options)
+ super().setUp()
def test_add_single_ip_address(self):
super().test_add_single_ip_address()
diff --git a/smoketest/scripts/cli/test_interfaces_pseudo_ethernet.py b/smoketest/scripts/cli/test_interfaces_pseudo_ethernet.py
index c1711c5a3..85e5e70bd 100755
--- a/smoketest/scripts/cli/test_interfaces_pseudo_ethernet.py
+++ b/smoketest/scripts/cli/test_interfaces_pseudo_ethernet.py
@@ -19,22 +19,19 @@ import unittest
from base_interfaces_test import BasicInterfaceTest
class PEthInterfaceTest(BasicInterfaceTest.BaseTest):
-
def setUp(self):
- super().setUp()
- self._base_path = ['interfaces', 'pseudo-ethernet']
-
self._test_ip = True
self._test_ipv6 = True
self._test_mtu = True
self._test_vlan = True
self._test_qinq = True
-
+ self._base_path = ['interfaces', 'pseudo-ethernet']
self._options = {
'peth0': ['source-interface eth1'],
'peth1': ['source-interface eth1'],
}
self._interfaces = list(self._options)
+ super().setUp()
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_interfaces_tunnel.py b/smoketest/scripts/cli/test_interfaces_tunnel.py
index 45679e280..f7b7f99ca 100755
--- a/smoketest/scripts/cli/test_interfaces_tunnel.py
+++ b/smoketest/scripts/cli/test_interfaces_tunnel.py
@@ -62,11 +62,8 @@ def tunnel_conf(interface):
class TunnelInterfaceTest(BasicInterfaceTest.BaseTest):
def setUp(self):
- super().setUp()
-
- self._base_path = ['interfaces', 'tunnel']
self._test_mtu = True
-
+ self._base_path = ['interfaces', 'tunnel']
self.local_v4 = '192.0.2.1'
self.local_v6 = '2001:db8::1'
@@ -79,6 +76,7 @@ class TunnelInterfaceTest(BasicInterfaceTest.BaseTest):
}
self._interfaces = list(self._options)
+ super().setUp()
def tearDown(self):
self.session.delete(['interfaces', 'dummy', source_if])
diff --git a/smoketest/scripts/cli/test_interfaces_vxlan.py b/smoketest/scripts/cli/test_interfaces_vxlan.py
index f41c180ad..a9b0fc5a1 100755
--- a/smoketest/scripts/cli/test_interfaces_vxlan.py
+++ b/smoketest/scripts/cli/test_interfaces_vxlan.py
@@ -21,8 +21,6 @@ from base_interfaces_test import BasicInterfaceTest
class VXLANInterfaceTest(BasicInterfaceTest.BaseTest):
def setUp(self):
- super().setUp()
-
self._test_mtu = True
self._base_path = ['interfaces', 'vxlan']
self._options = {
@@ -30,6 +28,7 @@ class VXLANInterfaceTest(BasicInterfaceTest.BaseTest):
'vxlan1': ['vni 20', 'group 239.1.1.1', 'source-interface eth0'],
}
self._interfaces = list(self._options)
+ super().setUp()
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_interfaces_wireless.py b/smoketest/scripts/cli/test_interfaces_wireless.py
index 9d2f4ea59..ffaa7d523 100755
--- a/smoketest/scripts/cli/test_interfaces_wireless.py
+++ b/smoketest/scripts/cli/test_interfaces_wireless.py
@@ -33,8 +33,6 @@ def get_config_value(interface, key):
class WirelessInterfaceTest(BasicInterfaceTest.BaseTest):
def setUp(self):
- super().setUp()
-
self._base_path = ['interfaces', 'wireless']
self._options = {
'wlan0': ['physical-device phy0', 'ssid VyOS-WIFI-0',
@@ -47,6 +45,7 @@ class WirelessInterfaceTest(BasicInterfaceTest.BaseTest):
'type access-point', 'address 192.0.2.13/30', 'channel 0'],
}
self._interfaces = list(self._options)
+ super().setUp()
def test_wireless_add_single_ip_address(self):
# derived method to check if member interfaces are enslaved properly
diff --git a/smoketest/scripts/cli/test_protocols_bgp.py b/smoketest/scripts/cli/test_protocols_bgp.py
index 984419e76..941d7828f 100755
--- a/smoketest/scripts/cli/test_protocols_bgp.py
+++ b/smoketest/scripts/cli/test_protocols_bgp.py
@@ -44,7 +44,8 @@ neighbor_config = {
'cap_strict' : '',
},
'192.0.2.3' : {
- 'description' : 'foo bar baz',
+# XXX: not available in current Perl backend
+# 'description' : 'foo bar baz',
'remote_as' : '200',
'passive' : '',
'multi_hop' : '5',
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-03 05:09:23 +0000