#!/usr/bin/env python
+# SPDX-License-Identifier: ISC
#
# Copyright (c) 2020 by Niral Networks, Inc. ("Niral Networks")
# Used Copyright (c) 2018 by Network Device Education Foundation,
# Inc. ("NetDEF") in this file.
#
-# Permission to use, copy, modify, and/or distribute this software
-# for any purpose with or without fee is hereby granted, provided
-# that the above copyright notice and this permission notice appear
-# in all copies.
-#
-# THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
-# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
-# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
-# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
-# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
-# WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
-# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
-# OF THIS SOFTWARE.
-#
"""
test_isis_topo1_vrf.py: Test ISIS vrf topology.
"""
-import collections
import functools
import json
import os
import re
import sys
import pytest
-import platform
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))
from lib.topotest import iproute2_is_vrf_capable
from lib.common_config import required_linux_kernel_version
-from mininet.topo import Topo
pytestmark = [pytest.mark.isisd]
]
-class ISISTopo1(Topo):
- "Simple two layer ISIS vrf topology"
-
- def build(self, *_args, **_opts):
- "Build function"
- tgen = get_topogen(self)
+def build_topo(tgen):
+ "Build function"
- # Add ISIS routers:
- # r1 r2
- # | sw1 | sw2
- # r3 r4
- # | |
- # sw3 sw4
- # \ /
- # r5
- for routern in range(1, 6):
- tgen.add_router("r{}".format(routern))
+ # Add ISIS routers:
+ # r1 r2
+ # | sw1 | sw2
+ # r3 r4
+ # | |
+ # sw3 sw4
+ # \ /
+ # r5
+ for routern in range(1, 6):
+ tgen.add_router("r{}".format(routern))
- # r1 <- sw1 -> r3
- sw = tgen.add_switch("sw1")
- sw.add_link(tgen.gears["r1"])
- sw.add_link(tgen.gears["r3"])
+ # r1 <- sw1 -> r3
+ sw = tgen.add_switch("sw1")
+ sw.add_link(tgen.gears["r1"])
+ sw.add_link(tgen.gears["r3"])
- # r2 <- sw2 -> r4
- sw = tgen.add_switch("sw2")
- sw.add_link(tgen.gears["r2"])
- sw.add_link(tgen.gears["r4"])
+ # r2 <- sw2 -> r4
+ sw = tgen.add_switch("sw2")
+ sw.add_link(tgen.gears["r2"])
+ sw.add_link(tgen.gears["r4"])
- # r3 <- sw3 -> r5
- sw = tgen.add_switch("sw3")
- sw.add_link(tgen.gears["r3"])
- sw.add_link(tgen.gears["r5"])
+ # r3 <- sw3 -> r5
+ sw = tgen.add_switch("sw3")
+ sw.add_link(tgen.gears["r3"])
+ sw.add_link(tgen.gears["r5"])
- # r4 <- sw4 -> r5
- sw = tgen.add_switch("sw4")
- sw.add_link(tgen.gears["r4"])
- sw.add_link(tgen.gears["r5"])
+ # r4 <- sw4 -> r5
+ sw = tgen.add_switch("sw4")
+ sw.add_link(tgen.gears["r4"])
+ sw.add_link(tgen.gears["r5"])
def setup_module(mod):
"Sets up the pytest environment"
- tgen = Topogen(ISISTopo1, mod.__name__)
+ tgen = Topogen(build_topo, mod.__name__)
tgen.start_topology()
logger.info("Testing with VRF Lite support")
"ip link add {0}-cust1 type vrf table 1001",
"ip link add loop1 type dummy",
"ip link set {0}-eth0 master {0}-cust1",
- "ip link set {0}-eth1 master {0}-cust1",
]
+ eth1_cmds = ["ip link set {0}-eth1 master {0}-cust1"]
+
# For all registered routers, load the zebra configuration file
for rname, router in tgen.routers().items():
# create VRF rx-cust1 and link rx-eth0 to rx-cust1
for cmd in cmds:
output = tgen.net[rname].cmd(cmd.format(rname))
+ # If router has an rX-eth1, link that to vrf also
+ if "{}-eth1".format(rname) in router.links.keys():
+ for cmd in eth1_cmds:
+ output = output + tgen.net[rname].cmd(cmd.format(rname))
+
for rname, router in tgen.routers().items():
router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
for rname, router in tgen.routers().items():
filename = "{0}/{1}/{1}_route.json".format(CWD, rname)
expected = json.loads(open(filename, "r").read())
- actual = router.vtysh_cmd(
- "show ip route vrf {0}-cust1 json".format(rname), isjson=True
- )
- assertmsg = "Router '{}' routes mismatch".format(rname)
- assert topotest.json_cmp(actual, expected) is None, assertmsg
+
+ def compare_routing_table(router, expected):
+ "Helper function to ensure zebra rib convergence"
+
+ actual = router.vtysh_cmd(
+ "show ip route vrf {0}-cust1 json".format(rname), isjson=True
+ )
+ return topotest.json_cmp(actual, expected)
+
+ test_func = functools.partial(compare_routing_table, router, expected)
+ (result, diff) = topotest.run_and_expect(test_func, None, count=20, wait=1)
+ assertmsg = "Router '{}' routes mismatch diff: {}".format(rname, diff)
+ assert result, assertmsg
def test_isis_linux_route_installation():
for rname, router in tgen.routers().items():
filename = "{0}/{1}/{1}_route6.json".format(CWD, rname)
expected = json.loads(open(filename, "r").read())
- actual = router.vtysh_cmd(
- "show ipv6 route vrf {}-cust1 json".format(rname), isjson=True
- )
- assertmsg = "Router '{}' routes mismatch".format(rname)
- assert topotest.json_cmp(actual, expected) is None, assertmsg
+ def compare_routing_table(router, expected):
+ "Helper function to ensure zebra rib convergence"
+ actual = router.vtysh_cmd(
+ "show ipv6 route vrf {}-cust1 json".format(rname), isjson=True
+ )
+ return topotest.json_cmp(actual, expected)
+
+ test_func = functools.partial(compare_routing_table, router, expected)
+ (result, diff) = topotest.run_and_expect(test_func, None, count=20, wait=1)
+ assertmsg = "Router '{}' routes mismatch diff: ".format(rname, diff)
+ assert result, assertmsg
def test_isis_linux_route6_installation():
https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
"""
for k, v in merge_dct.items():
- if (
- k in dct
- and isinstance(dct[k], dict)
- and isinstance(merge_dct[k], collections.Mapping)
- ):
+ if k in dct and isinstance(dct[k], dict) and topotest.is_mapping(merge_dct[k]):
dict_merge(dct[k], merge_dct[k])
else:
dct[k] = merge_dct[k]
projects / mirror_frr.git / blobdiff