diff options
Diffstat (limited to 'smoketest/scripts/cli/test_vrf.py')
| -rwxr-xr-x | smoketest/scripts/cli/test_vrf.py | 121 |
1 files changed, 119 insertions, 2 deletions
diff --git a/smoketest/scripts/cli/test_vrf.py b/smoketest/scripts/cli/test_vrf.py index 5270e758a..856baa070 100755 --- a/smoketest/scripts/cli/test_vrf.py +++ b/smoketest/scripts/cli/test_vrf.py @@ -16,19 +16,44 @@ import re import os +import json import unittest + from netifaces import interfaces from vyos.configsession import ConfigSession from vyos.configsession import ConfigSessionError +from vyos.ifconfig import Interface +from vyos.ifconfig import Section +from vyos.template import is_ipv6 +from vyos.util import cmd from vyos.util import read_file from vyos.validate import is_intf_addr_assigned -from vyos.ifconfig import Interface base_path = ['vrf'] vrfs = ['red', 'green', 'blue', 'foo-bar', 'baz_foo'] +def get_vrf_ipv4_routes(vrf): + return json.loads(cmd(f'ip -4 -j route show vrf {vrf}')) + +def get_vrf_ipv6_routes(vrf): + return json.loads(cmd(f'ip -6 -j route show vrf {vrf}')) + class VRFTest(unittest.TestCase): + _interfaces = [] + + @classmethod + def setUpClass(cls): + # we need to filter out VLAN interfaces identified by a dot (.) + # in their name - just in case! + if 'TEST_ETH' in os.environ: + tmp = os.environ['TEST_ETH'].split() + cls._interfaces = tmp + else: + for tmp in Section.interfaces('ethernet'): + if not '.' in tmp: + cls._interfaces.append(tmp) + def setUp(self): self.session = ConfigSession(os.getpid()) @@ -120,5 +145,97 @@ class VRFTest(unittest.TestCase): with self.assertRaises(ConfigSessionError): self.session.commit() + def test_vrf_assign_interface(self): + vrf = vrfs[0] + table = '5000' + self.session.set(['vrf', 'name', vrf, 'table', table]) + + for interface in self._interfaces: + section = Section.section(interface) + self.session.set(['interfaces', section, interface, 'vrf', vrf]) + + # commit changes + self.session.commit() + + # Verify & cleanup + for interface in self._interfaces: + # os.readlink resolves to: '../../../../../virtual/net/foovrf' + tmp = os.readlink(f'/sys/class/net/{interface}/master').split('/')[-1] + self.assertEqual(tmp, vrf) + # cleanup + section = Section.section(interface) + self.session.delete(['interfaces', section, interface, 'vrf']) + + def test_vrf_static_routes(self): + routes = { + '10.0.0.0/8' : { + 'next_hop' : '192.0.2.2', + 'distance' : '200', + 'next_hop_vrf' : 'default', + }, + '172.16.0.0/12' : { + 'next_hop' : '192.0.2.3', + 'next_hop_vrf' : 'default', + }, + '192.168.0.0/16' : { + 'next_hop' : '192.0.2.3', + }, + '2001:db8:1000::/48' : { + 'next_hop' : '2001:db8::2', + }, + } + + table = '2000' + for vrf in vrfs: + base = base_path + ['name', vrf] + self.session.set(base + ['table', str(table)]) + + # required interface for leaking to default table + self.session.set(['interfaces', 'ethernet', 'eth0', 'address', '192.0.2.1/24']) + + # we also need an interface in "UP" state to install routes + self.session.set(['interfaces', 'dummy', f'dum{table}', 'vrf', vrf]) + self.session.set(['interfaces', 'dummy', f'dum{table}', 'address', '192.0.2.1/24']) + self.session.set(['interfaces', 'dummy', f'dum{table}', 'address', '2001:db8::1/64']) + table = str(int(table) + 1) + + proto_base = ['protocols', 'vrf', vrf, 'static'] + for route, route_config in routes.items(): + route_type = 'route' + if is_ipv6(route): + route_type = 'route6' + self.session.set(proto_base + [route_type, route, 'next-hop', route_config['next_hop']]) + if 'distance' in route_config: + self.session.set(proto_base + [route_type, route, 'next-hop', route_config['next_hop'], 'distance', route_config['distance']]) + if 'next_hop_vrf' in route_config: + self.session.set(proto_base + [route_type, route, 'next-hop', route_config['next_hop'], 'next-hop-vrf', route_config['next_hop_vrf']]) + + # commit changes + self.session.commit() + + # Verify routes + table = '2000' + for vrf in vrfs: + for route, route_config in routes.items(): + if is_ipv6(route): + tmp = get_vrf_ipv6_routes(vrf) + else: + tmp = get_vrf_ipv4_routes(vrf) + + found = False + for result in tmp: + if 'dst' in result and result['dst'] == route: + if 'gateway' in result and result['gateway'] == route_config['next_hop']: + found = True + + self.assertTrue(found) + + # Cleanup + self.session.delete(['protocols', 'vrf', vrf]) + self.session.delete(['interfaces', 'dummy', f'dum{table}']) + self.session.delete(['interfaces', 'ethernet', 'eth0', 'address', '192.0.2.1/24']) + + table = str(int(table) + 1) + if __name__ == '__main__': - unittest.main(verbosity=2, failfast=True) + unittest.main(verbosity=2) |