diff options
Diffstat (limited to 'smoketest/scripts/cli/test_protocols_static.py')
| -rwxr-xr-x | smoketest/scripts/cli/test_protocols_static.py | 169 |
1 files changed, 103 insertions, 66 deletions
diff --git a/smoketest/scripts/cli/test_protocols_static.py b/smoketest/scripts/cli/test_protocols_static.py index cf591f060..0d3228cc7 100755 --- a/smoketest/scripts/cli/test_protocols_static.py +++ b/smoketest/scripts/cli/test_protocols_static.py @@ -14,23 +14,18 @@ # 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 base_vyostest_shim import VyOSUnitTestSHIM + from vyos.configsession import ConfigSession from vyos.configsession import ConfigSessionError from vyos.template import is_ipv6 -from vyos.util import cmd +from vyos.util import get_interface_config base_path = ['protocols', 'static'] vrf_path = ['protocols', 'vrf'] -def getFRRCconfig(vrf=None): - if vrf: - return cmd(f'vtysh -c "show run" | sed -n "/^vrf {vrf}/,/^!/p"') - else: - return cmd(f'vtysh -c "show run" | sed -n "/^ip route/,/^!/p"') - routes = { '10.0.0.0/8' : { 'next_hop' : { @@ -85,30 +80,27 @@ routes = { }, } -vrfs = ['red', 'green', 'blue'] tables = ['80', '81', '82'] -class StaticRouteTest(unittest.TestCase): +class TestProtocolsStatic(VyOSUnitTestSHIM.TestCase): def setUp(self): - self.session = ConfigSession(os.getpid()) + # This is our "target" VRF when leaking routes: + self.cli_set(['vrf', 'name', 'black', 'table', '43210']) def tearDown(self): for route, route_config in routes.items(): route_type = 'route' if is_ipv6(route): route_type = 'route6' - self.session.delete(base_path + [route_type, route]) - - for vrf in vrfs: - self.session.delete(vrf_path + [vrf]) + self.cli_delete(base_path + [route_type, route]) for table in tables: - self.session.delete(base_path + ['table', table]) + self.cli_delete(base_path + ['table', table]) - self.session.commit() - del self.session + tmp = self.getFRRconfig('', end='') + self.cli_commit() - def test_protocols_static(self): + def test_01_static(self): for route, route_config in routes.items(): route_type = 'route' if is_ipv6(route): @@ -116,39 +108,39 @@ class StaticRouteTest(unittest.TestCase): base = base_path + [route_type, route] if 'next_hop' in route_config: for next_hop, next_hop_config in route_config['next_hop'].items(): - self.session.set(base + ['next-hop', next_hop]) + self.cli_set(base + ['next-hop', next_hop]) if 'disable' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'disable']) + self.cli_set(base + ['next-hop', next_hop, 'disable']) if 'distance' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'distance', next_hop_config['distance']]) + self.cli_set(base + ['next-hop', next_hop, 'distance', next_hop_config['distance']]) if 'interface' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'interface', next_hop_config['interface']]) + self.cli_set(base + ['next-hop', next_hop, 'interface', next_hop_config['interface']]) if 'vrf' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'vrf', next_hop_config['vrf']]) + self.cli_set(base + ['next-hop', next_hop, 'vrf', next_hop_config['vrf']]) if 'interface' in route_config: for interface, interface_config in route_config['interface'].items(): - self.session.set(base + ['interface', interface]) + self.cli_set(base + ['interface', interface]) if 'disable' in interface_config: - self.session.set(base + ['interface', interface, 'disable']) + self.cli_set(base + ['interface', interface, 'disable']) if 'distance' in interface_config: - self.session.set(base + ['interface', interface, 'distance', interface_config['distance']]) + self.cli_set(base + ['interface', interface, 'distance', interface_config['distance']]) if 'vrf' in interface_config: - self.session.set(base + ['interface', interface, 'vrf', interface_config['vrf']]) + self.cli_set(base + ['interface', interface, 'vrf', interface_config['vrf']]) if 'blackhole' in route_config: - self.session.set(base + ['blackhole']) + self.cli_set(base + ['blackhole']) if 'distance' in route_config['blackhole']: - self.session.set(base + ['blackhole', 'distance', route_config['blackhole']['distance']]) + self.cli_set(base + ['blackhole', 'distance', route_config['blackhole']['distance']]) if 'tag' in route_config['blackhole']: - self.session.set(base + ['blackhole', 'tag', route_config['blackhole']['tag']]) + self.cli_set(base + ['blackhole', 'tag', route_config['blackhole']['tag']]) # commit changes - self.session.commit() + self.cli_commit() # Verify FRR bgpd configuration - frrconfig = getFRRCconfig() + frrconfig = self.getFRRconfig('ip route', end='') # Verify routes for route, route_config in routes.items(): @@ -195,7 +187,7 @@ class StaticRouteTest(unittest.TestCase): self.assertIn(tmp, frrconfig) - def test_protocols_static_table(self): + def test_02_static_table(self): for table in tables: for route, route_config in routes.items(): route_type = 'route' @@ -205,39 +197,39 @@ class StaticRouteTest(unittest.TestCase): if 'next_hop' in route_config: for next_hop, next_hop_config in route_config['next_hop'].items(): - self.session.set(base + ['next-hop', next_hop]) + self.cli_set(base + ['next-hop', next_hop]) if 'disable' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'disable']) + self.cli_set(base + ['next-hop', next_hop, 'disable']) if 'distance' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'distance', next_hop_config['distance']]) + self.cli_set(base + ['next-hop', next_hop, 'distance', next_hop_config['distance']]) if 'interface' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'interface', next_hop_config['interface']]) + self.cli_set(base + ['next-hop', next_hop, 'interface', next_hop_config['interface']]) if 'vrf' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'vrf', next_hop_config['vrf']]) + self.cli_set(base + ['next-hop', next_hop, 'vrf', next_hop_config['vrf']]) if 'interface' in route_config: for interface, interface_config in route_config['interface'].items(): - self.session.set(base + ['interface', interface]) + self.cli_set(base + ['interface', interface]) if 'disable' in interface_config: - self.session.set(base + ['interface', interface, 'disable']) + self.cli_set(base + ['interface', interface, 'disable']) if 'distance' in interface_config: - self.session.set(base + ['interface', interface, 'distance', interface_config['distance']]) + self.cli_set(base + ['interface', interface, 'distance', interface_config['distance']]) if 'vrf' in interface_config: - self.session.set(base + ['interface', interface, 'vrf', interface_config['vrf']]) + self.cli_set(base + ['interface', interface, 'vrf', interface_config['vrf']]) if 'blackhole' in route_config: - self.session.set(base + ['blackhole']) + self.cli_set(base + ['blackhole']) if 'distance' in route_config['blackhole']: - self.session.set(base + ['blackhole', 'distance', route_config['blackhole']['distance']]) + self.cli_set(base + ['blackhole', 'distance', route_config['blackhole']['distance']]) if 'tag' in route_config['blackhole']: - self.session.set(base + ['blackhole', 'tag', route_config['blackhole']['tag']]) + self.cli_set(base + ['blackhole', 'tag', route_config['blackhole']['tag']]) # commit changes - self.session.commit() + self.cli_commit() # Verify FRR bgpd configuration - frrconfig = getFRRCconfig() + frrconfig = self.getFRRconfig('ip route', end='') for table in tables: # Verify routes @@ -289,50 +281,68 @@ class StaticRouteTest(unittest.TestCase): self.assertIn(tmp, frrconfig) - def test_protocols_vrf_static(self): - for vrf in vrfs: + def test_03_static_vrf(self): + # Create VRF instances and apply the static routes from above to FRR. + # Re-read the configured routes and match them if they are programmed + # properly. This also includes VRF leaking + vrfs = { + 'red' : { 'table' : '1000' }, + 'green' : { 'table' : '2000' }, + 'blue' : { 'table' : '3000' }, + } + + for vrf, vrf_config in vrfs.items(): + vrf_base_path = ['vrf', 'name', vrf] + self.cli_set(vrf_base_path + ['table', vrf_config['table']]) + for route, route_config in routes.items(): route_type = 'route' if is_ipv6(route): route_type = 'route6' - base = vrf_path + [vrf, 'static', route_type, route] + route_base_path = vrf_base_path + ['protocols', 'static', route_type, route] if 'next_hop' in route_config: for next_hop, next_hop_config in route_config['next_hop'].items(): - self.session.set(base + ['next-hop', next_hop]) + self.cli_set(route_base_path + ['next-hop', next_hop]) if 'disable' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'disable']) + self.cli_set(route_base_path + ['next-hop', next_hop, 'disable']) if 'distance' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'distance', next_hop_config['distance']]) + self.cli_set(route_base_path + ['next-hop', next_hop, 'distance', next_hop_config['distance']]) if 'interface' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'interface', next_hop_config['interface']]) + self.cli_set(route_base_path + ['next-hop', next_hop, 'interface', next_hop_config['interface']]) if 'vrf' in next_hop_config: - self.session.set(base + ['next-hop', next_hop, 'vrf', next_hop_config['vrf']]) + self.cli_set(route_base_path + ['next-hop', next_hop, 'vrf', next_hop_config['vrf']]) if 'interface' in route_config: for interface, interface_config in route_config['interface'].items(): - self.session.set(base + ['interface', interface]) + self.cli_set(route_base_path + ['interface', interface]) if 'disable' in interface_config: - self.session.set(base + ['interface', interface, 'disable']) + self.cli_set(route_base_path + ['interface', interface, 'disable']) if 'distance' in interface_config: - self.session.set(base + ['interface', interface, 'distance', interface_config['distance']]) + self.cli_set(route_base_path + ['interface', interface, 'distance', interface_config['distance']]) if 'vrf' in interface_config: - self.session.set(base + ['interface', interface, 'vrf', interface_config['vrf']]) + self.cli_set(route_base_path + ['interface', interface, 'vrf', interface_config['vrf']]) if 'blackhole' in route_config: - self.session.set(base + ['blackhole']) + self.cli_set(route_base_path + ['blackhole']) if 'distance' in route_config['blackhole']: - self.session.set(base + ['blackhole', 'distance', route_config['blackhole']['distance']]) + self.cli_set(route_base_path + ['blackhole', 'distance', route_config['blackhole']['distance']]) if 'tag' in route_config['blackhole']: - self.session.set(base + ['blackhole', 'tag', route_config['blackhole']['tag']]) + self.cli_set(route_base_path + ['blackhole', 'tag', route_config['blackhole']['tag']]) # commit changes - self.session.commit() + self.cli_commit() + + for vrf, vrf_config in vrfs.items(): + tmp = get_interface_config(vrf) + + # Compare VRF table ID + self.assertEqual(tmp['linkinfo']['info_data']['table'], int(vrf_config['table'])) + self.assertEqual(tmp['linkinfo']['info_kind'], 'vrf') - for vrf in vrfs: # Verify FRR bgpd configuration - frrconfig = getFRRCconfig(vrf) + frrconfig = self.getFRRconfig(f'vrf {vrf}') self.assertIn(f'vrf {vrf}', frrconfig) # Verify routes @@ -380,6 +390,33 @@ class StaticRouteTest(unittest.TestCase): self.assertIn(tmp, frrconfig) + self.cli_delete(['vrf']) + + def test_04_static_zebra_route_map(self): + # Implemented because of T3328 + self.debug = True + route_map = 'foo-static-in' + self.cli_set(['policy', 'route-map', route_map, 'rule', '10', 'action', 'permit']) + + self.cli_set(base_path + ['route-map', route_map]) + # commit changes + self.cli_commit() + + # Verify FRR configuration + zebra_route_map = f'ip protocol static route-map {route_map}' + frrconfig = self.getFRRconfig(zebra_route_map) + self.assertIn(zebra_route_map, frrconfig) + + # Remove the route-map again + self.cli_delete(base_path + ['route-map']) + # commit changes + self.cli_commit() + + # Verify FRR configuration + frrconfig = self.getFRRconfig(zebra_route_map) + self.assertNotIn(zebra_route_map, frrconfig) + + self.cli_delete(['policy', 'route-map', route_map]) if __name__ == '__main__': unittest.main(verbosity=2) |