Merge pull request #16202 from y-bharath14/srib-topotest-d

tests: suppress unused variables at topotests
This commit is contained in:
Donatas Abraitis 2024-06-14 10:25:25 +03:00 committed by GitHub
commit e8ce036a7b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
98 changed files with 248 additions and 248 deletions

View File

@ -95,7 +95,7 @@ def expect_route_missing(router, iptype, route):
"show {} route json".format(iptype), "show {} route json".format(iptype),
{route: None}, {route: None},
) )
rv, result = topotest.run_and_expect(test_func, None, count=20, wait=1) _, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
assertmsg = '"{}" convergence failure'.format(router) assertmsg = '"{}" convergence failure'.format(router)
assert result is None, assertmsg assert result is None, assertmsg

View File

@ -61,7 +61,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
# check for zebra capability # check for zebra capability
for rname, router in router_list.items(): for _, router in router_list.items():
if router.check_capability(TopoRouter.RD_ZEBRA, "--vrfwnetns") == False: if router.check_capability(TopoRouter.RD_ZEBRA, "--vrfwnetns") == False:
return pytest.skip( return pytest.skip(
"Skipping BFD Topo1 VRF NETNS feature. VRF NETNS backend not available on FRR" "Skipping BFD Topo1 VRF NETNS feature. VRF NETNS backend not available on FRR"

View File

@ -68,7 +68,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -51,7 +51,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -49,7 +49,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -49,7 +49,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -89,12 +89,12 @@ def test_bgp_aggregate_address_origin():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge, router) test_func = functools.partial(_bgp_converge, router)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, 'Failed to see bgp convergence in "{}"'.format(router) assert result is None, 'Failed to see bgp convergence in "{}"'.format(router)
test_func = functools.partial(_bgp_aggregate_address_has_metric, router) test_func = functools.partial(_bgp_aggregate_address_has_metric, router)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert ( assert (
result is None result is None

View File

@ -52,7 +52,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -73,7 +73,7 @@ def test_bgp_aggregator_zero():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge) test_func = functools.partial(_bgp_converge)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, 'Failed bgp convergence in "{}"'.format(tgen.gears["r1"]) assert result is None, 'Failed bgp convergence in "{}"'.format(tgen.gears["r1"])
@ -99,7 +99,7 @@ def test_bgp_aggregator_zero():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_has_correct_aggregator_route_with_good_asn) test_func = functools.partial(_bgp_has_correct_aggregator_route_with_good_asn)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, 'Aggregator AS attribute not found in "{}"'.format( assert result is None, 'Aggregator AS attribute not found in "{}"'.format(
tgen.gears["r1"] tgen.gears["r1"]

View File

@ -73,7 +73,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -49,7 +49,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -48,7 +48,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -85,12 +85,12 @@ def test_bgp_as_wide_bgp_identifier():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge, tgen.gears["r1"]) test_func = functools.partial(_bgp_converge, tgen.gears["r1"])
success, result = topotest.run_and_expect(test_func, None, count=260, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=260, wait=0.5)
assert result is None, 'Failed to converge: "{}"'.format(tgen.gears["r1"]) assert result is None, 'Failed to converge: "{}"'.format(tgen.gears["r1"])
test_func = functools.partial(_bgp_failed, tgen.gears["r3"]) test_func = functools.partial(_bgp_failed, tgen.gears["r3"])
success, result = topotest.run_and_expect(test_func, None, count=260, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=260, wait=0.5)
assert result is None, 'Bad BGP Identifier notification not sent: "{}"'.format( assert result is None, 'Bad BGP Identifier notification not sent: "{}"'.format(
tgen.gears["r3"] tgen.gears["r3"]

View File

@ -60,7 +60,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -97,7 +97,7 @@ def test_bgp_asdot_regex():
logger.info("Check if neighbor sessions are up in {}".format(router1.name)) logger.info("Check if neighbor sessions are up in {}".format(router1.name))
test_func = partial(_bgp_converge, router1) test_func = partial(_bgp_converge, router1)
success, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5)
assert result is None, 'Failed to see BGP convergence in "{}"'.format(router1.name) assert result is None, 'Failed to see BGP convergence in "{}"'.format(router1.name)
logger.info("BGP neighbor session is up in {}".format(router1.name)) logger.info("BGP neighbor session is up in {}".format(router1.name))

View File

@ -73,7 +73,7 @@ def test_bgp_aggregator_zero():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge) test_func = functools.partial(_bgp_converge)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "More than one prefix seen at r1, SHOULD be only one." assert result is None, "More than one prefix seen at r1, SHOULD be only one."
def _bgp_has_correct_routes_without_asn_0(): def _bgp_has_correct_routes_without_asn_0():
@ -82,7 +82,7 @@ def test_bgp_aggregator_zero():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_has_correct_routes_without_asn_0) test_func = functools.partial(_bgp_has_correct_routes_without_asn_0)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Failed listing 192.168.100.101/32, SHOULD be accepted." assert result is None, "Failed listing 192.168.100.101/32, SHOULD be accepted."

View File

@ -156,7 +156,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
# For all registered routers, load the zebra configuration file # For all registered routers, load the zebra configuration file
for rname, router in router_list.items(): for _, router in router_list.items():
router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf") router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
router.load_config(TopoRouter.RD_OSPF, "") router.load_config(TopoRouter.RD_OSPF, "")
router.load_config(TopoRouter.RD_BGP, "") router.load_config(TopoRouter.RD_BGP, "")

View File

@ -156,7 +156,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
# For all registered routers, load the zebra configuration file # For all registered routers, load the zebra configuration file
for rname, router in router_list.items(): for _, router in router_list.items():
router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf") router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
router.load_config(TopoRouter.RD_OSPF, "") router.load_config(TopoRouter.RD_OSPF, "")
router.load_config(TopoRouter.RD_BGP, "") router.load_config(TopoRouter.RD_BGP, "")

View File

@ -155,7 +155,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
# For all registered routers, load the zebra configuration file # For all registered routers, load the zebra configuration file
for rname, router in router_list.items(): for _, router in router_list.items():
router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf") router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
router.load_config(TopoRouter.RD_OSPF, "") router.load_config(TopoRouter.RD_OSPF, "")
router.load_config(TopoRouter.RD_BGP, "") router.load_config(TopoRouter.RD_BGP, "")

View File

@ -155,7 +155,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
# For all registered routers, load the zebra configuration file # For all registered routers, load the zebra configuration file
for rname, router in router_list.items(): for _, router in router_list.items():
router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf") router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
router.load_config(TopoRouter.RD_OSPF, "") router.load_config(TopoRouter.RD_OSPF, "")
router.load_config(TopoRouter.RD_BGP, "") router.load_config(TopoRouter.RD_BGP, "")

View File

@ -45,7 +45,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -51,7 +51,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -125,14 +125,14 @@ def test_bgp_blackhole_community():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge) test_func = functools.partial(_bgp_converge)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, 'Failed bgp convergence in "{}"'.format(tgen.gears["r2"]) assert result is None, 'Failed bgp convergence in "{}"'.format(tgen.gears["r2"])
step("Check if 172.16.255.254/32 is not advertised to eBGP peers") step("Check if 172.16.255.254/32 is not advertised to eBGP peers")
test_func = functools.partial(_bgp_no_advertise_ebgp) test_func = functools.partial(_bgp_no_advertise_ebgp)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert ( assert (
result is None result is None
@ -142,7 +142,7 @@ def test_bgp_blackhole_community():
step("Check if 172.16.255.254/32 is advertised to iBGP peers") step("Check if 172.16.255.254/32 is advertised to iBGP peers")
test_func = functools.partial(_bgp_no_advertise_ibgp) test_func = functools.partial(_bgp_no_advertise_ibgp)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert ( assert (
result is None result is None
@ -152,7 +152,7 @@ def test_bgp_blackhole_community():
step("Verify if the nexthop set via route-map on r4 is marked valid") step("Verify if the nexthop set via route-map on r4 is marked valid")
test_func = functools.partial(_bgp_verify_nexthop_validity) test_func = functools.partial(_bgp_verify_nexthop_validity)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, 'Nexthops are not valid "{}"'.format(tgen.gears["r4"]) assert result is None, 'Nexthops are not valid "{}"'.format(tgen.gears["r4"])

View File

@ -48,7 +48,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -56,7 +56,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -40,7 +40,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -92,7 +92,7 @@ def test_bgp_community_alias():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge, router) test_func = functools.partial(_bgp_converge, router)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Cannot see BGP community aliases at r1" assert result is None, "Cannot see BGP community aliases at r1"
def _bgp_show_prefixes_by_alias(router): def _bgp_show_prefixes_by_alias(router):
@ -118,7 +118,7 @@ def test_bgp_community_alias():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_show_prefixes_by_alias, router) test_func = functools.partial(_bgp_show_prefixes_by_alias, router)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Cannot see BGP prefixes by community alias at r1" assert result is None, "Cannot see BGP prefixes by community alias at r1"
def _bgp_show_prefixes_by_large_community_list(router): def _bgp_show_prefixes_by_large_community_list(router):
@ -129,7 +129,7 @@ def test_bgp_community_alias():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_show_prefixes_by_large_community_list, router) test_func = functools.partial(_bgp_show_prefixes_by_large_community_list, router)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Cannot see BGP prefixes by large community list at r1" assert result is None, "Cannot see BGP prefixes by large community list at r1"

View File

@ -97,7 +97,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -133,7 +133,7 @@ def test_bgp_community_update_path_change():
step("Check if an initial topology is converged") step("Check if an initial topology is converged")
test_func = functools.partial(_bgp_converge_initial) test_func = functools.partial(_bgp_converge_initial)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Failed to see bgp convergence in c1" assert result is None, "Failed to see bgp convergence in c1"
step("Disable link between y1 and y2") step("Disable link between y1 and y2")
@ -146,7 +146,7 @@ def test_bgp_community_update_path_change():
step("Check if a topology is converged after a link down between y1 and y2") step("Check if a topology is converged after a link down between y1 and y2")
test_func = functools.partial(_bgp_converge_link_disabled) test_func = functools.partial(_bgp_converge_link_disabled)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Failed to see bgp convergence in y1" assert result is None, "Failed to see bgp convergence in y1"
def _bgp_check_for_duplicate_updates(): def _bgp_check_for_duplicate_updates():
@ -193,7 +193,7 @@ def test_bgp_community_update_path_change():
step("Check if a topology is converged after a link up between y1 and y2") step("Check if a topology is converged after a link up between y1 and y2")
test_func = functools.partial(_bgp_converge_link_enabled) test_func = functools.partial(_bgp_converge_link_enabled)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Failed to see bgp convergence in y1" assert result is None, "Failed to see bgp convergence in y1"
step( step(

View File

@ -159,7 +159,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -426,7 +426,7 @@ def test_bgp_conditional_advertisement_tc_2_1():
) )
test_func = functools.partial(exist_map_routes_present, router3) test_func = functools.partial(exist_map_routes_present, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = 'TC21: exist-map routes present in "router2" BGP table - ' msg = 'TC21: exist-map routes present in "router2" BGP table - '
assert result is None, msg + failed assert result is None, msg + failed
@ -455,7 +455,7 @@ def test_bgp_conditional_advertisement_tc_2_2():
) )
test_func = functools.partial(exist_map_routes_not_present, router3) test_func = functools.partial(exist_map_routes_not_present, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = 'TC22: exist-map routes not present in "router2" BGP table - ' msg = 'TC22: exist-map routes not present in "router2" BGP table - '
assert result is None, msg + failed assert result is None, msg + failed
@ -484,7 +484,7 @@ def test_bgp_conditional_advertisement_tc_2_3():
) )
test_func = functools.partial(default_route_withdrawn, router3) test_func = functools.partial(default_route_withdrawn, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC23: advertise-map with exist-map configuration is removed from peer - " msg = "TC23: advertise-map with exist-map configuration is removed from peer - "
assert result is None, msg + failed assert result is None, msg + failed
@ -513,7 +513,7 @@ def test_bgp_conditional_advertisement_tc_3_1():
) )
test_func = functools.partial(non_exist_map_routes_not_present, router3) test_func = functools.partial(non_exist_map_routes_not_present, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = 'TC31: non-exist-map routes not present in "router2" BGP table - ' msg = 'TC31: non-exist-map routes not present in "router2" BGP table - '
assert result is None, msg + failed assert result is None, msg + failed
@ -542,7 +542,7 @@ def test_bgp_conditional_advertisement_tc_3_2():
) )
test_func = functools.partial(non_exist_map_routes_present, router3) test_func = functools.partial(non_exist_map_routes_present, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = 'TC32: non-exist-map routes present in "router2" BGP table - ' msg = 'TC32: non-exist-map routes present in "router2" BGP table - '
assert result is None, msg + failed assert result is None, msg + failed
@ -571,7 +571,7 @@ def test_bgp_conditional_advertisement_tc_3_3():
) )
test_func = functools.partial(all_routes_advertised, router3) test_func = functools.partial(all_routes_advertised, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = ( msg = (
"TC33: advertise-map with non-exist-map configuration is removed from a peer - " "TC33: advertise-map with non-exist-map configuration is removed from a peer - "
@ -603,7 +603,7 @@ def test_bgp_conditional_advertisement_tc_4_1():
) )
test_func = functools.partial(non_exist_map_no_condition_route_map, router3) test_func = functools.partial(non_exist_map_no_condition_route_map, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = 'TC41: non-exist-map route-map removed in "router2" - ' msg = 'TC41: non-exist-map route-map removed in "router2" - '
assert result is None, msg + failed assert result is None, msg + failed
@ -632,7 +632,7 @@ def test_bgp_conditional_advertisement_tc_4_2():
) )
test_func = functools.partial(exist_map_no_condition_route_map, router3) test_func = functools.partial(exist_map_no_condition_route_map, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = 'TC42: exist-map route-map removed in "router2" - ' msg = 'TC42: exist-map route-map removed in "router2" - '
assert result is None, msg + failed assert result is None, msg + failed
@ -665,7 +665,7 @@ def test_bgp_conditional_advertisement_tc_5_1():
) )
test_func = functools.partial(exist_map_routes_present_rmap_filter, router3) test_func = functools.partial(exist_map_routes_present_rmap_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC51: exist-map routes present with route-map filter - " msg = "TC51: exist-map routes present with route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -694,7 +694,7 @@ def test_bgp_conditional_advertisement_tc_5_2():
) )
test_func = functools.partial(exist_map_routes_present_no_rmap_filter, router3) test_func = functools.partial(exist_map_routes_present_no_rmap_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC52: exist-map routes present, no route-map filter - " msg = "TC52: exist-map routes present, no route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -724,7 +724,7 @@ def test_bgp_conditional_advertisement_tc_5_3():
) )
test_func = functools.partial(non_exist_map_routes_present_rmap_filter, router3) test_func = functools.partial(non_exist_map_routes_present_rmap_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC53: non-exist-map routes present, with route-map filter - " msg = "TC53: non-exist-map routes present, with route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -753,7 +753,7 @@ def test_bgp_conditional_advertisement_tc_5_4():
) )
test_func = functools.partial(non_exist_map_routes_present_no_rmap_filter, router3) test_func = functools.partial(non_exist_map_routes_present_no_rmap_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC54: non-exist-map routes present, no route-map filter - " msg = "TC54: non-exist-map routes present, no route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -791,7 +791,7 @@ def test_bgp_conditional_advertisement_tc_6_1():
) )
test_func = functools.partial(exist_map_routes_not_present_rmap_filter, router3) test_func = functools.partial(exist_map_routes_not_present_rmap_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC61: exist-map routes not present, route-map filter - " msg = "TC61: exist-map routes not present, route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -820,7 +820,7 @@ def test_bgp_conditional_advertisement_tc_6_2():
) )
test_func = functools.partial(exist_map_routes_not_present_no_rmap_filter, router3) test_func = functools.partial(exist_map_routes_not_present_no_rmap_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC62: exist-map routes not present, no route-map filter - " msg = "TC62: exist-map routes not present, no route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -850,7 +850,7 @@ def test_bgp_conditional_advertisement_tc_6_3():
) )
test_func = functools.partial(non_exist_map_routes_not_present_rmap_filter, router3) test_func = functools.partial(non_exist_map_routes_not_present_rmap_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC63: non-exist-map routes not present, route-map filter - " msg = "TC63: non-exist-map routes not present, route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -881,7 +881,7 @@ def test_bgp_conditional_advertisement_tc_6_4():
test_func = functools.partial( test_func = functools.partial(
non_exist_map_routes_not_present_no_rmap_filter, router3 non_exist_map_routes_not_present_no_rmap_filter, router3
) )
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC64: non-exist-map routes not present, no route-map filter - " msg = "TC64: non-exist-map routes not present, no route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -919,7 +919,7 @@ def test_bgp_conditional_advertisement_tc_7_1():
) )
test_func = functools.partial(exist_map_routes_present_rmap2_filter, router3) test_func = functools.partial(exist_map_routes_present_rmap2_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC71: exist-map routes present, route-map filter - " msg = "TC71: exist-map routes present, route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -948,7 +948,7 @@ def test_bgp_conditional_advertisement_tc_7_2():
) )
test_func = functools.partial(exist_map_routes_present_no_rmap2_filter, router3) test_func = functools.partial(exist_map_routes_present_no_rmap2_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC72: exist-map routes present, no route-map filter - " msg = "TC72: exist-map routes present, no route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -978,7 +978,7 @@ def test_bgp_conditional_advertisement_tc_7_3():
) )
test_func = functools.partial(non_exist_map_routes_present_rmap2_filter, router3) test_func = functools.partial(non_exist_map_routes_present_rmap2_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC73: non-exist-map routes present, route-map filter - " msg = "TC73: non-exist-map routes present, route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -1007,7 +1007,7 @@ def test_bgp_conditional_advertisement_tc_7_4():
) )
test_func = functools.partial(non_exist_map_routes_present_no_rmap2_filter, router3) test_func = functools.partial(non_exist_map_routes_present_no_rmap2_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC74: non-exist-map routes present, no route-map filter - " msg = "TC74: non-exist-map routes present, no route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -1045,7 +1045,7 @@ def test_bgp_conditional_advertisement_tc_8_1():
) )
test_func = functools.partial(exist_map_routes_not_present_rmap2_filter, router3) test_func = functools.partial(exist_map_routes_not_present_rmap2_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC81: exist-map routes not present, route-map filter - " msg = "TC81: exist-map routes not present, route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -1074,7 +1074,7 @@ def test_bgp_conditional_advertisement_tc_8_2():
) )
test_func = functools.partial(exist_map_routes_not_present_no_rmap2_filter, router3) test_func = functools.partial(exist_map_routes_not_present_no_rmap2_filter, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC82: exist-map routes not present, no route-map filter - " msg = "TC82: exist-map routes not present, no route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -1106,7 +1106,7 @@ def test_bgp_conditional_advertisement_tc_8_3():
test_func = functools.partial( test_func = functools.partial(
non_exist_map_routes_not_present_rmap2_filter, router3 non_exist_map_routes_not_present_rmap2_filter, router3
) )
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC83: non-exist-map routes not present, route-map filter - " msg = "TC83: non-exist-map routes not present, route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -1137,7 +1137,7 @@ def test_bgp_conditional_advertisement_tc_8_4():
test_func = functools.partial( test_func = functools.partial(
non_exist_map_routes_not_present_no_rmap2_filter, router3 non_exist_map_routes_not_present_no_rmap2_filter, router3
) )
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC84: non-exist-map routes not present, no route-map filter - " msg = "TC84: non-exist-map routes not present, no route-map filter - "
assert result is None, msg + failed assert result is None, msg + failed
@ -1175,7 +1175,7 @@ def test_bgp_conditional_advertisement_tc_9_1():
) )
test_func = functools.partial(exist_map_routes_present_rmap2_network, router3) test_func = functools.partial(exist_map_routes_present_rmap2_network, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC91: exist-map routes present, route-map filter and network - " msg = "TC91: exist-map routes present, route-map filter and network - "
assert result is None, msg + failed assert result is None, msg + failed
@ -1204,7 +1204,7 @@ def test_bgp_conditional_advertisement_tc_9_2():
) )
test_func = functools.partial(exist_map_routes_present_rmap2_no_network, router3) test_func = functools.partial(exist_map_routes_present_rmap2_no_network, router3)
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC92: exist-map routes present, route-map filter and no network - " msg = "TC92: exist-map routes present, route-map filter and no network - "
assert result is None, msg + failed assert result is None, msg + failed
@ -1244,7 +1244,7 @@ def test_bgp_conditional_advertisement_tc_9_3():
test_func = functools.partial( test_func = functools.partial(
non_exist_map_routes_not_present_rmap2_network, router3 non_exist_map_routes_not_present_rmap2_network, router3
) )
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC93: non-exist-map routes not present, route-map filter and network - " msg = "TC93: non-exist-map routes not present, route-map filter and network - "
assert result is None, msg + failed assert result is None, msg + failed
@ -1275,7 +1275,7 @@ def test_bgp_conditional_advertisement_tc_9_4():
test_func = functools.partial( test_func = functools.partial(
non_exist_map_routes_not_present_rmap2_no_network, router3 non_exist_map_routes_not_present_rmap2_no_network, router3
) )
success, result = topotest.run_and_expect(test_func, None, count=90, wait=1) _, result = topotest.run_and_expect(test_func, None, count=90, wait=1)
msg = "TC94: non-exist-map routes not present, route-map filter and no network - " msg = "TC94: non-exist-map routes not present, route-map filter and no network - "
assert result is None, msg + failed assert result is None, msg + failed

View File

@ -51,7 +51,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -35,7 +35,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -48,7 +48,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -50,7 +50,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -49,7 +49,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -40,7 +40,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -85,19 +85,19 @@ def test_bgp_default_originate_route_map():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_check_if_received) test_func = functools.partial(_bgp_check_if_received)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "No 0.0.0.0/0 at r2 from r1" assert result is None, "No 0.0.0.0/0 at r2 from r1"
test_func = functools.partial(_bgp_check_if_originated) test_func = functools.partial(_bgp_check_if_originated)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "No 0.0.0.0/0 from r1 to r2" assert result is None, "No 0.0.0.0/0 from r1 to r2"
test_func = functools.partial(_bgp_route_is_valid, tgen.gears["r2"], "0.0.0.0/0") test_func = functools.partial(_bgp_route_is_valid, tgen.gears["r2"], "0.0.0.0/0")
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "Failed to see 0.0.0.0/0 in r2" assert result is None, "Failed to see 0.0.0.0/0 in r2"
test_func = functools.partial(_bgp_route_is_valid, tgen.gears["r2"], "0.0.0.0/1") test_func = functools.partial(_bgp_route_is_valid, tgen.gears["r2"], "0.0.0.0/1")
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "Failed to see 0.0.0.0/1 in r2" assert result is None, "Failed to see 0.0.0.0/1 in r2"

View File

@ -40,7 +40,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -80,12 +80,12 @@ def test_bgp_default_originate_route_map():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge, router) test_func = functools.partial(_bgp_converge, router)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, 'Failed to see bgp convergence in "{}"'.format(router) assert result is None, 'Failed to see bgp convergence in "{}"'.format(router)
test_func = functools.partial(_bgp_default_route_is_valid, router) test_func = functools.partial(_bgp_default_route_is_valid, router)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert ( assert (
result is None result is None

View File

@ -43,7 +43,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -84,7 +84,7 @@ def test_bgp_default_originate_route_map():
step("Converge network") step("Converge network")
test_func = functools.partial(_bgp_converge, router) test_func = functools.partial(_bgp_converge, router)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "Failed to see bgp convergence at r2" assert result is None, "Failed to see bgp convergence at r2"
step("Withdraw 10.0.0.0/22 from R2") step("Withdraw 10.0.0.0/22 from R2")
@ -94,7 +94,7 @@ def test_bgp_default_originate_route_map():
step("Check if we don't have 0.0.0.0/0 at R2") step("Check if we don't have 0.0.0.0/0 at R2")
test_func = functools.partial(_bgp_default_route_is_valid, router) test_func = functools.partial(_bgp_default_route_is_valid, router)
success, result = topotest.run_and_expect(test_func, not None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, not None, count=30, wait=0.5)
assert result is not None, "0.0.0.0/0 exists at r2" assert result is not None, "0.0.0.0/0 exists at r2"
step("Announce 10.0.0.0/22 from R2") step("Announce 10.0.0.0/22 from R2")
@ -102,7 +102,7 @@ def test_bgp_default_originate_route_map():
step("Check if we have 0.0.0.0/0 at R2") step("Check if we have 0.0.0.0/0 at R2")
test_func = functools.partial(_bgp_default_route_is_valid, router) test_func = functools.partial(_bgp_default_route_is_valid, router)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "0.0.0.0/0 does not exist at r2" assert result is None, "0.0.0.0/0 does not exist at r2"
step("Withdraw 10.0.0.0/22 from R2 again") step("Withdraw 10.0.0.0/22 from R2 again")
@ -112,7 +112,7 @@ def test_bgp_default_originate_route_map():
step("Check if we don't have 0.0.0.0/0 at R2 again") step("Check if we don't have 0.0.0.0/0 at R2 again")
test_func = functools.partial(_bgp_default_route_is_valid, router) test_func = functools.partial(_bgp_default_route_is_valid, router)
success, result = topotest.run_and_expect(test_func, not None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, not None, count=30, wait=0.5)
assert result is not None, "0.0.0.0/0 exists at r2" assert result is not None, "0.0.0.0/0 exists at r2"

View File

@ -42,7 +42,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -90,12 +90,12 @@ def test_bgp_default_originate_route_map():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge, router) test_func = functools.partial(_bgp_converge, router)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, 'Failed to see bgp convergence in "{}"'.format(router) assert result is None, 'Failed to see bgp convergence in "{}"'.format(router)
test_func = functools.partial(_bgp_default_route_has_metric, router) test_func = functools.partial(_bgp_default_route_has_metric, router)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert ( assert (
result is None result is None

View File

@ -41,7 +41,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -46,7 +46,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -98,7 +98,7 @@ def test_bgp_disable_addpath_rx():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(check_bgp_advertised_routes, r2) test_func = functools.partial(check_bgp_advertised_routes, r2)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "AddPath TX not working." assert result is None, "AddPath TX not working."
step("Check if AddPath RX is disabled on r1 and we receive only 2 paths.") step("Check if AddPath RX is disabled on r1 and we receive only 2 paths.")
@ -120,7 +120,7 @@ def test_bgp_disable_addpath_rx():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(check_bgp_disabled_addpath_rx, r1) test_func = functools.partial(check_bgp_disabled_addpath_rx, r1)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "AddPath RX advertised, but should not." assert result is None, "AddPath RX advertised, but should not."

View File

@ -51,7 +51,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -101,14 +101,14 @@ def test_bgp_maximum_prefix_invalid():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge, router) test_func = functools.partial(_bgp_converge, router)
success, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5)
assert result is None, 'Failed to see BGP convergence in "{}"'.format(router) assert result is None, 'Failed to see BGP convergence in "{}"'.format(router)
_bgp_distance_change(router) _bgp_distance_change(router)
test_func = functools.partial(_bgp_check_distance_change, router) test_func = functools.partial(_bgp_check_distance_change, router)
success, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5)
assert result is None, 'Failed to see applied BGP distance in RIB "{}"'.format( assert result is None, 'Failed to see applied BGP distance in RIB "{}"'.format(
router router

View File

@ -34,7 +34,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -52,7 +52,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -90,7 +90,7 @@ def test_bgp_ebgp_common_subnet_nh_unchanged():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge, r3) test_func = functools.partial(_bgp_converge, r3)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, 'Failed bgp convergence in "{}"'.format(r3) assert result is None, 'Failed bgp convergence in "{}"'.format(r3)
@ -100,7 +100,7 @@ def test_bgp_ebgp_common_subnet_nh_unchanged():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_nh_unchanged, r2) test_func = functools.partial(_bgp_nh_unchanged, r2)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, 'Wrong next-hop in "{}"'.format(r2) assert result is None, 'Wrong next-hop in "{}"'.format(r2)

View File

@ -71,7 +71,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -125,31 +125,31 @@ def test_ebgp_requires_policy():
# Scenario 1. # Scenario 1.
logger.info("Scenario 1: r2 receives 192.168.255.1/32 from r1") logger.info("Scenario 1: r2 receives 192.168.255.1/32 from r1")
test_func = functools.partial(_bgp_converge, "r2") test_func = functools.partial(_bgp_converge, "r2")
success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
assert success is True, "Failed bgp convergence (r2)" assert success is True, "Failed bgp convergence (r2)"
test_func = functools.partial(_bgp_has_routes, "r2") test_func = functools.partial(_bgp_has_routes, "r2")
success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
assert success is True, "r2 does not receive 192.168.255.1/32" assert success is True, "r2 does not receive 192.168.255.1/32"
# Scenario 2. # Scenario 2.
logger.info("Scenario 2: r3 must not send 192.168.255.1/32 to r4") logger.info("Scenario 2: r3 must not send 192.168.255.1/32 to r4")
test_func = functools.partial(_bgp_converge, "r4") test_func = functools.partial(_bgp_converge, "r4")
success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
assert success is True, "Failed bgp convergence (r4)" assert success is True, "Failed bgp convergence (r4)"
test_func = functools.partial(_bgp_advertised_routes, "r3") test_func = functools.partial(_bgp_advertised_routes, "r3")
success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
assert success is True, "r3 announced 192.168.255.1/32 to r4" assert success is True, "r3 announced 192.168.255.1/32 to r4"
# Scenario 3. # Scenario 3.
logger.info("Scenario 3: r6 receives 192.168.255.1/32 from r5 (iBGP)") logger.info("Scenario 3: r6 receives 192.168.255.1/32 from r5 (iBGP)")
test_func = functools.partial(_bgp_converge, "r6") test_func = functools.partial(_bgp_converge, "r6")
success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
assert success is True, "Failed bgp convergence (r6)" assert success is True, "Failed bgp convergence (r6)"
test_func = functools.partial(_bgp_has_routes, "r6") test_func = functools.partial(_bgp_has_routes, "r6")
success, result = topotest.run_and_expect(test_func, None, count=120, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=120, wait=0.5)
assert success is True, "r6 does not receive 192.168.255.1/32" assert success is True, "r6 does not receive 192.168.255.1/32"

View File

@ -47,7 +47,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -34,7 +34,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -50,7 +50,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -108,7 +108,7 @@ def setup_module(mod):
r2.run("ip link set r2-eth3 master vrf1") r2.run("ip link set r2-eth3 master vrf1")
# Load FRR config and initialize all routers # Load FRR config and initialize all routers
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -45,7 +45,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -41,7 +41,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -56,7 +56,7 @@ def setup_module(mod):
) )
tgen.gears["r{}".format(rtr)].run("sysctl -w net.mpls.conf.vrf1.input=1") tgen.gears["r{}".format(rtr)].run("sysctl -w net.mpls.conf.vrf1.input=1")
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -57,7 +57,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -42,7 +42,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -61,7 +61,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -65,7 +65,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -46,7 +46,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -63,7 +63,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -50,7 +50,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -142,7 +142,7 @@ def test_converge_bgplu():
"192.168.2.2/32", "192.168.2.2/32",
"0", "0",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r1, prefix 192.168.2.2/32 from r2 not present" assert success, "r1, prefix 192.168.2.2/32 from r2 not present"
# Check r2 gets prefix 192.168.2.1/32 # Check r2 gets prefix 192.168.2.1/32
@ -153,7 +153,7 @@ def test_converge_bgplu():
"192.168.2.1/32", "192.168.2.1/32",
"0", "0",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r2, prefix 192.168.2.1/32 from r1 not present" assert success, "r2, prefix 192.168.2.1/32 from r1 not present"
@ -178,7 +178,7 @@ def test_traffic_connectivity():
tgen = get_topogen() tgen = get_topogen()
func = functools.partial(_check_ping, "r1", "192.168.2.2", "192.168.2.1") func = functools.partial(_check_ping, "r1", "192.168.2.2", "192.168.2.1")
# tgen.mininet_cli() # tgen.mininet_cli()
success, result = topotest.run_and_expect(func, None, count=10, wait=0.5) _, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
assert result is None, "r1, ping to 192.168.2.2 from 192.168.2.1 fails" assert result is None, "r1, ping to 192.168.2.2 from 192.168.2.1 fails"

View File

@ -43,7 +43,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -82,17 +82,17 @@ def test_bgp_max_med_on_startup():
# Check session is established # Check session is established
test_func = functools.partial(_bgp_converge, router2) test_func = functools.partial(_bgp_converge, router2)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "Failed bgp convergence on r2" assert result is None, "Failed bgp convergence on r2"
# Check metric has value of max-med # Check metric has value of max-med
test_func = functools.partial(_bgp_has_routes, router2, 777) test_func = functools.partial(_bgp_has_routes, router2, 777)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "r2 does not receive routes with metric 777" assert result is None, "r2 does not receive routes with metric 777"
# Check that when the max-med timer expires, metric is updated # Check that when the max-med timer expires, metric is updated
test_func = functools.partial(_bgp_has_routes, router2, 0) test_func = functools.partial(_bgp_has_routes, router2, 0)
success, result = topotest.run_and_expect(test_func, None, count=16, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=16, wait=0.5)
assert result is None, "r2 does not receive routes with metric 0" assert result is None, "r2 does not receive routes with metric 0"

View File

@ -50,7 +50,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -45,7 +45,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -177,7 +177,7 @@ def test_bgp_maximum_prefix_out():
router1.vtysh_cmd(cmd) router1.vtysh_cmd(cmd)
test_func = functools.partial(_bgp_converge, router2, exp_prfxs) test_func = functools.partial(_bgp_converge, router2, exp_prfxs)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, 'Failed bgp convergence in "{}"'.format(router2) assert result is None, 'Failed bgp convergence in "{}"'.format(router2)

View File

@ -40,7 +40,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -76,7 +76,7 @@ def test_bgp_minimum_holdtime():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_neighbor_check_if_notification_sent) test_func = functools.partial(_bgp_neighbor_check_if_notification_sent)
success, result = topotest.run_and_expect(test_func, None, count=40, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=40, wait=0.5)
assert result is None, "Failed to send notification message\n" assert result is None, "Failed to send notification message\n"

View File

@ -46,7 +46,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_frr_config(os.path.join(CWD, "{}/frr.conf".format(rname))) router.load_frr_config(os.path.join(CWD, "{}/frr.conf".format(rname)))
tgen.start_router() tgen.start_router()

View File

@ -37,7 +37,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -57,7 +57,7 @@ def setup_module(mod):
tgen.gears["r2"].cmd("ip address add dev vrf1 192.0.2.8/32") tgen.gears["r2"].cmd("ip address add dev vrf1 192.0.2.8/32")
tgen.gears["r3"].cmd("ip address add dev vrf1 192.0.2.8/32") tgen.gears["r3"].cmd("ip address add dev vrf1 192.0.2.8/32")
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -34,7 +34,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -47,7 +47,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -33,7 +33,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -93,11 +93,11 @@ def test_r1_receive_and_advertise_prefix_sid_type1():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_check_type1_r1, router, "3.0.0.1/32", 800001, 1) test_func = functools.partial(_check_type1_r1, router, "3.0.0.1/32", 800001, 1)
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert result is None, 'Failed _check_type1_r1 in "{}"'.format(router) assert result is None, 'Failed _check_type1_r1 in "{}"'.format(router)
test_func = functools.partial(_check_type1_r1, router, "3.0.0.2/32", 800002, 2) test_func = functools.partial(_check_type1_r1, router, "3.0.0.2/32", 800002, 2)
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert result is None, 'Failed _check_type1_r1 in "{}"'.format(router) assert result is None, 'Failed _check_type1_r1 in "{}"'.format(router)
@ -160,11 +160,11 @@ def test_peer2_receive_prefix_sid_type1():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_check_type1_peer2, "3.0.0.1/32", label=8001) test_func = functools.partial(_check_type1_peer2, "3.0.0.1/32", label=8001)
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert result is None, 'Failed _check_type1_peer2 in "{}"'.format("peer2") assert result is None, 'Failed _check_type1_peer2 in "{}"'.format("peer2")
test_func = functools.partial(_check_type1_peer2, "3.0.0.2/32", label=8002) test_func = functools.partial(_check_type1_peer2, "3.0.0.2/32", label=8002)
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert result is None, 'Failed _check_type1_peer2 in "{}"'.format("peer2") assert result is None, 'Failed _check_type1_peer2 in "{}"'.format("peer2")

View File

@ -56,7 +56,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -114,19 +114,19 @@ def test_bgp_reject_as_sets():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge, router) test_func = functools.partial(_bgp_converge, router)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, 'Failed bgp convergence in "{}"'.format(router) assert result is None, 'Failed bgp convergence in "{}"'.format(router)
test_func = functools.partial( test_func = functools.partial(
_bgp_has_aggregated_route_with_stripped_as_set, router _bgp_has_aggregated_route_with_stripped_as_set, router
) )
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, 'Failed to see an aggregated route in "{}"'.format(router) assert result is None, 'Failed to see an aggregated route in "{}"'.format(router)
test_func = functools.partial(_bgp_announce_route_without_as_sets, router) test_func = functools.partial(_bgp_announce_route_without_as_sets, router)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert ( assert (
result is None result is None

View File

@ -109,7 +109,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, f"{rname}/zebra.conf") TopoRouter.RD_ZEBRA, os.path.join(CWD, f"{rname}/zebra.conf")
) )

View File

@ -41,7 +41,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_frr_config(os.path.join(CWD, "{}/frr.conf".format(rname))) router.load_frr_config(os.path.join(CWD, "{}/frr.conf".format(rname)))
tgen.start_router() tgen.start_router()

View File

@ -43,7 +43,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -86,7 +86,7 @@ def test_bgp_extcommunity_none():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge, router) test_func = functools.partial(_bgp_converge, router)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "BGP Converge failed" assert result is None, "BGP Converge failed"
def _bgp_extcommunity_strip(router): def _bgp_extcommunity_strip(router):
@ -111,7 +111,7 @@ def test_bgp_extcommunity_none():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_extcommunity_strip, router) test_func = functools.partial(_bgp_extcommunity_strip, router)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Failed to strip incoming extended communities from r2" assert result is None, "Failed to strip incoming extended communities from r2"

View File

@ -82,7 +82,7 @@ def test_correct_pair(tgen):
check_r2_established = functools.partial( check_r2_established = functools.partial(
check_session_established, router, neighbor_ip check_session_established, router, neighbor_ip
) )
success, result = topotest.run_and_expect( success, _ = topotest.run_and_expect(
check_r2_established, True, count=20, wait=3 check_r2_established, True, count=20, wait=3
) )
assert success, "Session with r2 is not Established" assert success, "Session with r2 is not Established"
@ -100,7 +100,7 @@ def test_role_pair_mismatch(tgen):
router = tgen.gears["r3"] router = tgen.gears["r3"]
neighbor_ip = "192.168.3.1" neighbor_ip = "192.168.3.1"
check_r3_mismatch = functools.partial(check_role_mismatch, router, neighbor_ip) check_r3_mismatch = functools.partial(check_role_mismatch, router, neighbor_ip)
success, result = topotest.run_and_expect(check_r3_mismatch, True, count=20, wait=3) success, _ = topotest.run_and_expect(check_r3_mismatch, True, count=20, wait=3)
assert success, "Session between r1 and r3 was not correctly closed" assert success, "Session between r1 and r3 was not correctly closed"
@ -111,7 +111,7 @@ def test_single_role_advertising(tgen):
check_r4_established = functools.partial( check_r4_established = functools.partial(
check_session_established, router, neighbor_ip check_session_established, router, neighbor_ip
) )
success, result = topotest.run_and_expect( success, _ = topotest.run_and_expect(
check_r4_established, True, count=20, wait=3 check_r4_established, True, count=20, wait=3
) )
assert success, "Session with r4 is not Established" assert success, "Session with r4 is not Established"
@ -129,7 +129,7 @@ def test_single_role_receiving(tgen):
check_r1_established = functools.partial( check_r1_established = functools.partial(
check_session_established, router, neighbor_ip check_session_established, router, neighbor_ip
) )
success, result = topotest.run_and_expect( success, _ = topotest.run_and_expect(
check_r1_established, True, count=20, wait=3 check_r1_established, True, count=20, wait=3
) )
assert success, "Session with r1 is not Established" assert success, "Session with r1 is not Established"
@ -145,7 +145,7 @@ def test_role_strict_mode(tgen):
router = tgen.gears["r5"] router = tgen.gears["r5"]
neighbor_ip = "192.168.5.1" neighbor_ip = "192.168.5.1"
check_r5_mismatch = functools.partial(check_role_mismatch, router, neighbor_ip) check_r5_mismatch = functools.partial(check_role_mismatch, router, neighbor_ip)
success, result = topotest.run_and_expect(check_r5_mismatch, True, count=20, wait=3) success, _ = topotest.run_and_expect(check_r5_mismatch, True, count=20, wait=3)
assert success, "Session between r1 and r5 was not correctly closed" assert success, "Session between r1 and r5 was not correctly closed"

View File

@ -69,7 +69,7 @@ def test_r10_routes(tgen):
] ]
return output == expected return output == expected
success, result = topotest.run_and_expect( success, _ = topotest.run_and_expect(
_routes_half_converged, True, count=20, wait=3 _routes_half_converged, True, count=20, wait=3
) )
assert success, "Routes did not converged" assert success, "Routes did not converged"

View File

@ -32,7 +32,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -41,7 +41,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -97,7 +97,7 @@ def test_bgp_route_map_match_ipv6_next_hop_access_list():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge, router) test_func = functools.partial(_bgp_converge, router)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Can't match routes using ipv6 next-hop access-list" assert result is None, "Can't match routes using ipv6 next-hop access-list"

View File

@ -46,7 +46,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_frr_config(os.path.join(CWD, "{}/frr.conf".format(rname))) router.load_frr_config(os.path.join(CWD, "{}/frr.conf".format(rname)))
tgen.start_router() tgen.start_router()

View File

@ -42,7 +42,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -83,12 +83,12 @@ def test_bgp_route_map_on_match_next():
# Check thst session is established # Check thst session is established
test_func = functools.partial(_bgp_converge, router2) test_func = functools.partial(_bgp_converge, router2)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "Failed bgp convergence on r2" assert result is None, "Failed bgp convergence on r2"
# Check that metric is 0 and weight is 100 for the received prefix # Check that metric is 0 and weight is 100 for the received prefix
test_func = functools.partial(_bgp_has_routes, router2, 0, 100) test_func = functools.partial(_bgp_has_routes, router2, 0, 100)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "r2 does not receive routes with metric 0 and weight 100" assert result is None, "r2 does not receive routes with metric 0 and weight 100"
# Update the route-map and add "on-match next" to entry 10 # Update the route-map and add "on-match next" to entry 10
@ -102,7 +102,7 @@ def test_bgp_route_map_on_match_next():
# Check that metric is 20 and weight is 100 for the received prefix # Check that metric is 20 and weight is 100 for the received prefix
test_func = functools.partial(_bgp_has_routes, router2, 20, 100) test_func = functools.partial(_bgp_has_routes, router2, 20, 100)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "r2 does not receive routes with metric 20 and weight 100" assert result is None, "r2 does not receive routes with metric 20 and weight 100"

View File

@ -61,7 +61,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -44,7 +44,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -44,7 +44,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -50,7 +50,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -46,7 +46,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -48,7 +48,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -47,7 +47,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -100,12 +100,12 @@ def test_bgp_set_local_preference():
return topotest.json_cmp(output, expected) return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_converge, router) test_func = functools.partial(_bgp_converge, router)
success, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5)
assert result is None, 'Failed to see BGP convergence in "{}"'.format(router) assert result is None, 'Failed to see BGP convergence in "{}"'.format(router)
test_func = functools.partial(_bgp_check_local_preference, router) test_func = functools.partial(_bgp_check_local_preference, router)
success, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5)
assert result is None, 'Failed to see applied BGP local-preference in "{}"'.format( assert result is None, 'Failed to see applied BGP local-preference in "{}"'.format(
router router

View File

@ -34,7 +34,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -73,7 +73,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -106,7 +106,7 @@ def setup_module(mod):
tgen = Topogen(build_topo, mod.__name__) tgen = Topogen(build_topo, mod.__name__)
tgen.start_topology() tgen.start_topology()
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(tgen.routers().items(), 1): for _, (rname, router) in enumerate(tgen.routers().items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -180,7 +180,7 @@ def check_rib(name, cmd, expected_file):
logger.info('[+] check {} "{}" {}'.format(name, cmd, expected_file)) logger.info('[+] check {} "{}" {}'.format(name, cmd, expected_file))
tgen = get_topogen() tgen = get_topogen()
func = functools.partial(_check, name, cmd, expected_file) func = functools.partial(_check, name, cmd, expected_file)
success, result = topotest.run_and_expect(func, None, count=10, wait=0.5) _, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
assert result is None, "Failed" assert result is None, "Failed"

View File

@ -147,7 +147,7 @@ def check_rib(name, cmd, expected_file):
logger.info('[+] check {} "{}" {}'.format(name, cmd, expected_file)) logger.info('[+] check {} "{}" {}'.format(name, cmd, expected_file))
tgen = get_topogen() tgen = get_topogen()
func = functools.partial(_check, name, cmd, expected_file) func = functools.partial(_check, name, cmd, expected_file)
success, result = topotest.run_and_expect(func, None, count=10, wait=0.5) _, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
assert result is None, "Failed" assert result is None, "Failed"

View File

@ -109,7 +109,7 @@ def check_rib(name, cmd, expected_file):
logger.info('[+] check {} "{}" {}'.format(name, cmd, expected_file)) logger.info('[+] check {} "{}" {}'.format(name, cmd, expected_file))
tgen = get_topogen() tgen = get_topogen()
func = functools.partial(_check, name, cmd, expected_file) func = functools.partial(_check, name, cmd, expected_file)
success, result = topotest.run_and_expect(func, None, count=10, wait=0.5) _, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
assert result is None, "Failed" assert result is None, "Failed"

View File

@ -106,7 +106,7 @@ def check_rib(name, cmd, expected_file):
logger.info('[+] check {} "{}" {}'.format(name, cmd, expected_file)) logger.info('[+] check {} "{}" {}'.format(name, cmd, expected_file))
tgen = get_topogen() tgen = get_topogen()
func = functools.partial(_check, name, cmd, expected_file) func = functools.partial(_check, name, cmd, expected_file)
success, result = topotest.run_and_expect(func, None, count=10, wait=0.5) _, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
assert result is None, "Failed" assert result is None, "Failed"

View File

@ -47,7 +47,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -52,7 +52,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -108,7 +108,7 @@ def test_bgp_tcp_mss():
logger.info("Check if neighbor sessions are up in {}".format(router1.name)) logger.info("Check if neighbor sessions are up in {}".format(router1.name))
test_func = functools.partial(_bgp_converge, router1) test_func = functools.partial(_bgp_converge, router1)
success, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5)
assert result is None, 'Failed to see BGP convergence in "{}"'.format(router1.name) assert result is None, 'Failed to see BGP convergence in "{}"'.format(router1.name)
logger.info("BGP neighbor session is up in {}".format(router1.name)) logger.info("BGP neighbor session is up in {}".format(router1.name))
@ -129,7 +129,7 @@ def test_bgp_tcp_mss():
"Check if neighbor session is up after reset in {}".format(router1.name) "Check if neighbor session is up after reset in {}".format(router1.name)
) )
test_func = functools.partial(_bgp_converge, router1) test_func = functools.partial(_bgp_converge, router1)
success, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=15, wait=0.5)
assert result is None, 'Failed to see BGP convergence after reset in "{}"'.format( assert result is None, 'Failed to see BGP convergence after reset in "{}"'.format(
router1.name router1.name
) )
@ -138,7 +138,7 @@ def test_bgp_tcp_mss():
"Verify if TCP MSS value is synced with neighbor in {}".format(router1.name) "Verify if TCP MSS value is synced with neighbor in {}".format(router1.name)
) )
test_func = functools.partial(_bgp_check_neighbor_tcp_mss, router1, "192.168.255.2") test_func = functools.partial(_bgp_check_neighbor_tcp_mss, router1, "192.168.255.2")
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert ( assert (
result is None result is None
), 'Failed to sync TCP MSS value over BGP session in "{}"'.format(router1.name) ), 'Failed to sync TCP MSS value over BGP session in "{}"'.format(router1.name)
@ -148,7 +148,7 @@ def test_bgp_tcp_mss():
"Verify if TCP MSS value is synced with neighbor in {}".format(router2.name) "Verify if TCP MSS value is synced with neighbor in {}".format(router2.name)
) )
test_func = functools.partial(_bgp_check_neighbor_tcp_mss, router2, "192.168.255.1") test_func = functools.partial(_bgp_check_neighbor_tcp_mss, router2, "192.168.255.1")
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) _, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert ( assert (
result is None result is None
), 'Failed to sync TCP MSS value over BGP session in "{}"'.format(router2.name) ), 'Failed to sync TCP MSS value over BGP session in "{}"'.format(router2.name)

View File

@ -41,7 +41,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_frr_config(os.path.join(CWD, "{}/frr.conf".format(rname))) router.load_frr_config(os.path.join(CWD, "{}/frr.conf".format(rname)))
tgen.start_router() tgen.start_router()

View File

@ -42,7 +42,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -84,7 +84,7 @@ def test_bgp_unnumbered_removal():
step("Ensure Convergence of BGP") step("Ensure Convergence of BGP")
test_func = functools.partial(_bgp_converge) test_func = functools.partial(_bgp_converge)
success, result = topotest.run_and_expect(test_func, None, count=60, wait=1) _, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
assert result is None, 'Failed bgp convergence in "{}"'.format(tgen.gears["r2"]) assert result is None, 'Failed bgp convergence in "{}"'.format(tgen.gears["r2"])
@ -109,7 +109,7 @@ def test_bgp_unnumbered_removal():
step("Ensure that BGP does not crash") step("Ensure that BGP does not crash")
test_func = functools.partial(_bgp_nexthop_cache) test_func = functools.partial(_bgp_nexthop_cache)
success, result = topotest.run_and_expect(test_func, True, count=10, wait=1) _, result = topotest.run_and_expect(test_func, True, count=10, wait=1)
assert result is True, "BGP did not crash on r1" assert result is True, "BGP did not crash on r1"

View File

@ -90,7 +90,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )
@ -169,7 +169,7 @@ def test_bgp_update_delay():
# Check r2 initial convergence in default table # Check r2 initial convergence in default table
test_func = functools.partial(_bgp_converge, router2) test_func = functools.partial(_bgp_converge, router2)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=1) _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assert result is None, 'Failed bgp convergence in "{}"'.format(router2) assert result is None, 'Failed bgp convergence in "{}"'.format(router2)
@ -195,7 +195,7 @@ def test_bgp_update_delay():
router2.vtysh_cmd("""clear ip bgp *""") router2.vtysh_cmd("""clear ip bgp *""")
test_func = functools.partial(_bgp_check_update_delay_in_progress, router2) test_func = functools.partial(_bgp_check_update_delay_in_progress, router2)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=1) _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assert result is None, 'Failed to set update-delay max-delay timer "{}"'.format( assert result is None, 'Failed to set update-delay max-delay timer "{}"'.format(
router2 router2
@ -203,7 +203,7 @@ def test_bgp_update_delay():
# Check that r2 only installs route learned from r4 after the max-delay timer expires # Check that r2 only installs route learned from r4 after the max-delay timer expires
test_func = functools.partial(_bgp_check_route_install, router2) test_func = functools.partial(_bgp_check_route_install, router2)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=1) _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assert result is None, 'Failed to install route after update-delay "{}"'.format( assert result is None, 'Failed to install route after update-delay "{}"'.format(
router2 router2
@ -219,7 +219,7 @@ def test_bgp_update_delay():
) )
test_func = functools.partial(_bgp_check_update_delay_and_wait, router2) test_func = functools.partial(_bgp_check_update_delay_and_wait, router2)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=1) _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assert ( assert (
result is None result is None
@ -229,7 +229,7 @@ def test_bgp_update_delay():
router2.vtysh_cmd("""clear ip bgp *""") router2.vtysh_cmd("""clear ip bgp *""")
test_func = functools.partial(_bgp_check_route_install, router3) test_func = functools.partial(_bgp_check_route_install, router3)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=1) _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assert ( assert (
result is None result is None
@ -250,7 +250,7 @@ def test_bgp_update_delay():
router2.vtysh_cmd("""clear ip bgp *""") router2.vtysh_cmd("""clear ip bgp *""")
test_func = functools.partial(_bgp_check_route_install, router2) test_func = functools.partial(_bgp_check_route_install, router2)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=1) _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assert result is None, 'Failed to remove update-delay delay timing "{}"'.format( assert result is None, 'Failed to remove update-delay delay timing "{}"'.format(
router2 router2
@ -266,14 +266,14 @@ def test_bgp_update_delay():
# Check that r2 default instance and vrf1 have the max-delay and establish set # Check that r2 default instance and vrf1 have the max-delay and establish set
test_func = functools.partial(_bgp_check_update_delay_and_wait, router2) test_func = functools.partial(_bgp_check_update_delay_and_wait, router2)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=1) _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assert result is None, 'Failed to set update-delay in default instance "{}"'.format( assert result is None, 'Failed to set update-delay in default instance "{}"'.format(
router2 router2
) )
test_func = functools.partial(_bgp_check_vrf_update_delay_and_wait, router2) test_func = functools.partial(_bgp_check_vrf_update_delay_and_wait, router2)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=1) _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assert result is None, 'Failed to set update-delay in vrf1 "{}"'.format(router2) assert result is None, 'Failed to set update-delay in vrf1 "{}"'.format(router2)
@ -281,7 +281,7 @@ def test_bgp_update_delay():
router2.vtysh_cmd("""clear ip bgp *""") router2.vtysh_cmd("""clear ip bgp *""")
test_func = functools.partial(_bgp_check_route_install, router3) test_func = functools.partial(_bgp_check_route_install, router3)
success, result = topotest.run_and_expect(test_func, None, count=30, wait=1) _, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assert ( assert (
result is None result is None

View File

@ -67,7 +67,7 @@ def setup_module(mod):
router_list = tgen.routers() router_list = tgen.routers()
for i, (rname, router) in enumerate(router_list.items(), 1): for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config( router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
) )

View File

@ -203,7 +203,7 @@ def bgp_vpnv4_prefix_check(router, rd, prefix, label, nexthop):
) )
func = functools.partial(_check, router, prefix, rd, label, nexthop) func = functools.partial(_check, router, prefix, rd, label, nexthop)
success, result = topotest.run_and_expect(func, None, count=20, wait=0.5) _, result = topotest.run_and_expect(func, None, count=20, wait=0.5)
assert_msg = "{}, show bgp ipv4 vpn {}, rd {}, label {} nexthop {}".format( assert_msg = "{}, show bgp ipv4 vpn {}, rd {}, label {} nexthop {}".format(
router.name, prefix, rd, label, nexthop router.name, prefix, rd, label, nexthop
) )
@ -324,7 +324,7 @@ def check_show_bgp_vpn_ok(router, vpnv4_entries):
for prefix, rname_to_test in vpnv4_entries.items(): for prefix, rname_to_test in vpnv4_entries.items():
func = functools.partial(_check_nexthop_available, router, prefix) func = functools.partial(_check_nexthop_available, router, prefix)
success, result = topotest.run_and_expect(func, None, count=20, wait=0.5) _, result = topotest.run_and_expect(func, None, count=20, wait=0.5)
assert result is None, "Failed to detect prefix {} on router {}".format( assert result is None, "Failed to detect prefix {} on router {}".format(
prefix, router.name prefix, router.name
) )
@ -456,7 +456,7 @@ def test_r3_prefixes_removed():
prefix, prefix,
"444:3", "444:3",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {} still present".format(router.name, prefix) assert success, "{}, vpnv4 update {} still present".format(router.name, prefix)
# diagnostic # diagnostic
@ -491,7 +491,7 @@ def test_r3_prefixes_removed():
prefix, prefix,
"444:3", "444:3",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {} still present".format(router.name, prefix) assert success, "{}, vpnv4 update {} still present".format(router.name, prefix)
logger.info( logger.info(
@ -502,7 +502,7 @@ def test_r3_prefixes_removed():
test_func = functools.partial( test_func = functools.partial(
check_show_mpls_table_entry_label_not_found, router, label_ip_entries[prefix] check_show_mpls_table_entry_label_not_found, router, label_ip_entries[prefix]
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r1, mpls entry with in_label {} still present".format( assert success, "r1, mpls entry with in_label {} still present".format(
label_ip_entries[prefix] label_ip_entries[prefix]
) )
@ -536,7 +536,7 @@ def test_r3_prefixes_added_back():
prefix, prefix,
"444:3", "444:3",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {} not present".format(router.name, prefix) assert success, "{}, vpnv4 update {} not present".format(router.name, prefix)
logger.info( logger.info(
@ -564,7 +564,7 @@ def test_r3_prefixes_added_back():
prefix, prefix,
"444:3", "444:3",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {} not present".format(router.name, prefix) assert success, "{}, vpnv4 update {} not present".format(router.name, prefix)
# diagnostic # diagnostic
@ -611,7 +611,7 @@ def test_unconfigure_nexthop_change_nexthop_self():
test_func = functools.partial( test_func = functools.partial(
check_show_mpls_table_entry_label_not_found, router, label check_show_mpls_table_entry_label_not_found, router, label
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r1, mpls entry for {} with in_label {} still present".format( assert success, "r1, mpls entry for {} with in_label {} still present".format(
prefix, label prefix, label
) )
@ -626,7 +626,7 @@ def test_unconfigure_nexthop_change_nexthop_self():
"444:3", "444:3",
label=label, label=label,
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, mpls vpn update {} label {} is present".format( assert success, "{}, mpls vpn update {} label {} is present".format(
router.name, prefix, label router.name, prefix, label
) )
@ -639,7 +639,7 @@ def test_unconfigure_nexthop_change_nexthop_self():
"444:3", "444:3",
nexthop="192.168.1.3", nexthop="192.168.1.3",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, mpls vpn update {} label {} is present".format( assert success, "{}, mpls vpn update {} label {} is present".format(
router.name, prefix, label router.name, prefix, label
) )
@ -727,7 +727,7 @@ def test_declare_vpn_network_with_different_label():
label=label, label=label,
nexthop="192.168.1.3", nexthop="192.168.1.3",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {}, label {} not present".format( assert success, "{}, vpnv4 update {}, label {} not present".format(
router.name, prefix, label router.name, prefix, label
) )
@ -787,7 +787,7 @@ def test_filter_vpn_network_from_r1():
"172.31.0.0/24", "172.31.0.0/24",
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {}, is still present".format( assert success, "{}, vpnv4 update {}, is still present".format(
router.name, prefix router.name, prefix
) )
@ -804,7 +804,7 @@ def test_filter_vpn_network_from_r1():
test_func = functools.partial( test_func = functools.partial(
check_show_mpls_table_entry_label_not_found, router, int(label) check_show_mpls_table_entry_label_not_found, router, int(label)
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r1, mpls entry for {} with in_label {} still present".format( assert success, "r1, mpls entry for {} with in_label {} still present".format(
prefix, label prefix, label
) )
@ -833,7 +833,7 @@ def test_unfilter_vpn_network_from_r1():
test_func = functools.partial( test_func = functools.partial(
check_show_bgp_vpn_prefix_found, router, "ipv4", prefix, "444:1" check_show_bgp_vpn_prefix_found, router, "ipv4", prefix, "444:1"
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {}, is not present".format(router.name, prefix) assert success, "{}, vpnv4 update {}, is not present".format(router.name, prefix)
vpnv4_checks = { vpnv4_checks = {

View File

@ -245,7 +245,7 @@ router bgp 65500 vrf vrf1
prefix, prefix,
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {} still present".format(router.name, prefix) assert success, "{}, vpnv4 update {} still present".format(router.name, prefix)
@ -279,7 +279,7 @@ route-map RMAP permit 1
prefix, prefix,
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {} still not present".format(router.name, prefix) assert success, "{}, vpnv4 update {} still not present".format(router.name, prefix)
@ -309,7 +309,7 @@ route-map RMAP permit 1
prefix, prefix,
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {} still present".format(router.name, prefix) assert success, "{}, vpnv4 update {} still present".format(router.name, prefix)
@ -339,7 +339,7 @@ router bgp 65500 vrf vrf1
prefix, prefix,
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {} still not present".format(router.name, prefix) assert success, "{}, vpnv4 update {} still not present".format(router.name, prefix)
@ -371,7 +371,7 @@ route-map RMAP permit 1
prefix, prefix,
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {} still present".format(router.name, prefix) assert success, "{}, vpnv4 update {} still present".format(router.name, prefix)
@ -400,7 +400,7 @@ route-map RMAP permit 1
prefix, prefix,
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, vpnv4 update {} still not present".format(router.name, prefix) assert success, "{}, vpnv4 update {} still not present".format(router.name, prefix)
@ -482,7 +482,7 @@ router bgp 65501
expected, expected,
exact=True, exact=True,
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r2, vpnv4 update 172.31.0.1 still present" assert success, "r2, vpnv4 update 172.31.0.1 still present"

View File

@ -174,7 +174,7 @@ def bgp_vpnv4_table_check(router, group, label_list=None, label_value_expected=N
stored_label_inited = False stored_label_inited = False
for prefix in group: for prefix in group:
test_func = functools.partial(check_bgp_vpnv4_prefix_presence, router, prefix) test_func = functools.partial(check_bgp_vpnv4_prefix_presence, router, prefix)
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, prefix ipv4 vpn {} is not installed yet".format( assert success, "{}, prefix ipv4 vpn {} is not installed yet".format(
router.name, prefix router.name, prefix
) )
@ -305,7 +305,7 @@ def mpls_table_check(router, blacklist=None, label_list=None, whitelist=None):
test_func = functools.partial( test_func = functools.partial(
check_show_mpls_table, router, blacklist, label_list, whitelist check_show_mpls_table, router, blacklist, label_list, whitelist
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, MPLS labels check fail: {}".format(router.name, result) assert success, "{}, MPLS labels check fail: {}".format(router.name, result)
@ -450,7 +450,7 @@ def test_flapping_bgp_vrf_down():
test_func = functools.partial( test_func = functools.partial(
_bgp_prefix_not_found, tgen.gears["r1"], "vrf1", "ipv4", "172.31.0.11/32" _bgp_prefix_not_found, tgen.gears["r1"], "vrf1", "ipv4", "172.31.0.11/32"
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert ( assert (
success success
), "r1, prefix 172.31.0.11/32 from r11 did not disappear. r11 still connected to rr ?" ), "r1, prefix 172.31.0.11/32 from r11 did not disappear. r11 still connected to rr ?"
@ -488,7 +488,7 @@ def test_flapping_bgp_vrf_up():
"172.31.0.11/32", "172.31.0.11/32",
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert ( assert (
success success
), "r2, prefix 172.31.0.11/32 from r11 not present. r11 still disconnected from rr ?" ), "r2, prefix 172.31.0.11/32 from r11 not present. r11 still disconnected from rr ?"
@ -518,7 +518,7 @@ def test_recursive_route():
"172.31.0.30/32", "172.31.0.30/32",
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r2, vpnv4 update 172.31.0.30 not found" assert success, "r2, vpnv4 update 172.31.0.30 not found"
bgp_vpnv4_table_check(tgen.gears["r2"], group=PREFIXES_R11 + ["172.31.0.30/32"]) bgp_vpnv4_table_check(tgen.gears["r2"], group=PREFIXES_R11 + ["172.31.0.30/32"])
@ -544,7 +544,7 @@ def test_recursive_route():
"172.31.0.30/32", "172.31.0.30/32",
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r2, vpnv4 update 172.31.0.30 still present" assert success, "r2, vpnv4 update 172.31.0.30 still present"
@ -570,7 +570,7 @@ def test_prefix_changes_interface():
"172.31.0.50/32", "172.31.0.50/32",
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r2, vpnv4 update 172.31.0.50 not found" assert success, "r2, vpnv4 update 172.31.0.50 not found"
# diagnostic # diagnostic
@ -616,7 +616,7 @@ def test_prefix_changes_interface():
"444:1", "444:1",
label=oldlabel, label=oldlabel,
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert ( assert (
success success
), "r2, vpnv4 update 172.31.0.50 with old label {0} still present".format(oldlabel) ), "r2, vpnv4 update 172.31.0.50 with old label {0} still present".format(oldlabel)
@ -633,7 +633,7 @@ def test_prefix_changes_interface():
"172.31.0.50/32", "172.31.0.50/32",
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r2, vpnv4 update 172.31.0.50 not found" assert success, "r2, vpnv4 update 172.31.0.50 not found"
label_list = set() label_list = set()
@ -699,7 +699,7 @@ def test_changing_default_label_value():
test_func = functools.partial( test_func = functools.partial(
check_show_mpls_table_entry_label_found, router, 222, "vrf1" check_show_mpls_table_entry_label_found, router, 222, "vrf1"
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r1, mpls entry with label 222 not found" assert success, "r1, mpls entry with label 222 not found"
# check label repartition is ok # check label repartition is ok
@ -746,7 +746,7 @@ def test_unconfigure_allocation_mode_nexthop():
test_func = functools.partial( test_func = functools.partial(
check_show_mpls_table_entry_label_not_found, router, 17 check_show_mpls_table_entry_label_not_found, router, 17
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r1, mpls entry with label 17 still present" assert success, "r1, mpls entry with label 17 still present"
# Check vpnv4 routes from r1 # Check vpnv4 routes from r1
@ -794,7 +794,7 @@ def test_reconfigure_allocation_mode_nexthop():
test_func = functools.partial( test_func = functools.partial(
check_show_mpls_table_entry_label_not_found, router, 17 check_show_mpls_table_entry_label_not_found, router, 17
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r1, mpls entry with label 17 still present" assert success, "r1, mpls entry with label 17 still present"
# Check vpnv4 routes from r1 # Check vpnv4 routes from r1

View File

@ -173,7 +173,7 @@ def bgp_vpnv6_table_check(router, group, label_list=None, label_value_expected=N
stored_label_inited = False stored_label_inited = False
for prefix in group: for prefix in group:
test_func = functools.partial(check_bgp_vpnv6_prefix_presence, router, prefix) test_func = functools.partial(check_bgp_vpnv6_prefix_presence, router, prefix)
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, prefix ipv6 vpn {} is not installed yet".format( assert success, "{}, prefix ipv6 vpn {} is not installed yet".format(
router.name, prefix router.name, prefix
) )
@ -300,7 +300,7 @@ def mpls_table_check(router, blacklist=None, label_list=None, whitelist=None):
test_func = functools.partial( test_func = functools.partial(
check_show_mpls_table, router, blacklist, label_list, whitelist check_show_mpls_table, router, blacklist, label_list, whitelist
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "{}, MPLS labels check fail: {}".format(router.name, result) assert success, "{}, MPLS labels check fail: {}".format(router.name, result)
@ -446,7 +446,7 @@ def test_flapping_bgp_vrf_down():
test_func = functools.partial( test_func = functools.partial(
_bgp_prefix_not_found, tgen.gears["r1"], "vrf1", "ipv6", "172:31::11/128" _bgp_prefix_not_found, tgen.gears["r1"], "vrf1", "ipv6", "172:31::11/128"
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert ( assert (
success success
), "r1, prefix 172:31::11/128 from r11 did not disappear. r11 still connected to rr ?" ), "r1, prefix 172:31::11/128 from r11 did not disappear. r11 still connected to rr ?"
@ -488,7 +488,7 @@ def test_flapping_bgp_vrf_up():
"172:31::11/128", "172:31::11/128",
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert ( assert (
success success
), "r2, prefix 172:31::11/128 from r11 not present. r11 still disconnected from rr ?" ), "r2, prefix 172:31::11/128 from r11 not present. r11 still disconnected from rr ?"
@ -526,7 +526,7 @@ def test_recursive_route():
# Check r2 received vpnv6 update with 172:31::30 # Check r2 received vpnv6 update with 172:31::30
test_func = functools.partial(_prefix30_found, tgen.gears["r2"]) test_func = functools.partial(_prefix30_found, tgen.gears["r2"])
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r2, VPNv6 update 172:31::30 not found" assert success, "r2, VPNv6 update 172:31::30 not found"
# that route should be sent along with label for 192::2:11 # that route should be sent along with label for 192::2:11
@ -549,7 +549,7 @@ def test_recursive_route():
# Check r2 removed 172:31::30 vpnv6 update # Check r2 removed 172:31::30 vpnv6 update
test_func = functools.partial(_prefix30_not_found, tgen.gears["r2"]) test_func = functools.partial(_prefix30_not_found, tgen.gears["r2"])
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r2, VPNv6 update 172:31::30 still present" assert success, "r2, VPNv6 update 172:31::30 still present"
@ -575,7 +575,7 @@ def test_prefix_changes_interface():
"172:31::50/128", "172:31::50/128",
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r2, VPNv6 update 172:31::50 not found" assert success, "r2, VPNv6 update 172:31::50 not found"
# diagnostic # diagnostic
@ -621,7 +621,7 @@ def test_prefix_changes_interface():
"444:1", "444:1",
label=oldlabel, label=oldlabel,
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert ( assert (
success success
), "r2, vpnv6 update 172:31::50 with old label {0} still present".format(oldlabel) ), "r2, vpnv6 update 172:31::50 with old label {0} still present".format(oldlabel)
@ -638,7 +638,7 @@ def test_prefix_changes_interface():
"172:31::50/128", "172:31::50/128",
"444:1", "444:1",
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r2, vpnv6 update 172:31::50 not found" assert success, "r2, vpnv6 update 172:31::50 not found"
label_list = set() label_list = set()
@ -704,7 +704,7 @@ def test_changing_default_label_value():
test_func = functools.partial( test_func = functools.partial(
check_show_mpls_table_entry_label_found, router, 222, "vrf1" check_show_mpls_table_entry_label_found, router, 222, "vrf1"
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r1, mpls entry with label 222 not found" assert success, "r1, mpls entry with label 222 not found"
# check label repartition is ok # check label repartition is ok
@ -750,7 +750,7 @@ def test_unconfigure_allocation_mode_nexthop():
test_func = functools.partial( test_func = functools.partial(
check_show_mpls_table_entry_label_not_found, router, 17 check_show_mpls_table_entry_label_not_found, router, 17
) )
success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5) success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assert success, "r1, mpls entry with label 17 still present" assert success, "r1, mpls entry with label 17 still present"
# Check vpnv6 routes from r1 # Check vpnv6 routes from r1