import pytest
import re
import time
-from time import sleep
# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
for rtrNum in range(1, 6):
tgen.add_router("r{}".format(rtrNum))
- # create ExaBGP peers
- for peer_num in range(1, 5):
- tgen.add_exabgp_peer(
- "peer{}".format(peer_num),
- ip="192.168.101.{}".format(peer_num + 2),
- defaultRoute="via 192.168.101.1",
- )
-
# Setup Switches and connections
for swNum in range(1, 11):
tgen.add_switch("sw{}".format(swNum))
tgen.gears["r2"].add_link(tgen.gears["sw5"])
tgen.gears["r5"].add_link(tgen.gears["sw5"])
- # Add ExaBGP peers to sw4
- tgen.gears["peer1"].add_link(tgen.gears["sw4"])
- tgen.gears["peer2"].add_link(tgen.gears["sw4"])
- tgen.gears["peer3"].add_link(tgen.gears["sw4"])
- tgen.gears["peer4"].add_link(tgen.gears["sw4"])
-
#####################################################
#
# end test_bgp_delayopen_dual
-def test_bgp_dampening_setup():
- "BGP route-flap dampening test setup"
-
- # This test starts four ExaBGP peers, adds them as neighbors to the
- # configuration of router r1 and checks if connections get established.
-
- tgen = get_topogen()
-
- # Skip if previous fatal error condition is raised
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info("Starting BGP route-flap dampening test setup")
-
- # Start ExaBGP peers connected to r1 via switch 4
- logger.info("Starting ExaBGP peers")
- for peer_num in range(1, 5):
- logger.info("Creating named pipe for ExaBGP peer peer{}".format(peer_num))
- fifo_in = "/var/run/exabgp_peer{}.in".format(peer_num)
- if os.path.exists(fifo_in):
- os.remove(fifo_in)
- os.mkfifo(fifo_in, 0o777)
- logger.info("Starting ExaBGP on peer peer{}".format(peer_num))
- peer = tgen.gears["peer{}".format(peer_num)]
- peer_dir = os.path.join(CWD, "peer{}".format(peer_num))
- env_file = os.path.join(CWD, "exabgp.env")
- peer.start(peer_dir, env_file)
-
- # Add ExaBGP peers to configuration of router r2
- logger.info("Adding ExaBGP peers as neighbors to configuration of router r2")
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "neighbor 192.168.101.3 remote-as 65403"'
- )
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "neighbor 192.168.101.3 route-map testmap-in"'
- )
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "neighbor 192.168.101.3 route-map testmap-out"'
- )
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "neighbor 192.168.101.4 remote-as 65404"'
- )
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "neighbor 192.168.101.4 route-map testmap-in"'
- )
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "neighbor 192.168.101.4 route-map testmap-out"'
- )
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "neighbor 192.168.101.5 remote-as 65405"'
- )
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "neighbor 192.168.101.5 route-map testmap-in"'
- )
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "neighbor 192.168.101.5 route-map testmap-out"'
- )
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "neighbor 192.168.101.6 remote-as 65406"'
- )
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "neighbor 192.168.101.6 route-map testmap-in"'
- )
- tgen.net["r1"].cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "neighbor 192.168.101.6 route-map testmap-out"'
- )
-
- # Check if exabgp peers are up and running
- logger.info("Checking for established connections to ExaBGP peers on router r1")
- router = tgen.gears["r1"]
- reffile = os.path.join(CWD, "r1/bgp_damp_setup.json")
- expected = json.loads(open(reffile).read())
- test_func = functools.partial(
- topotest.router_json_cmp, router, "show ip bgp summary json", expected
- )
- _, res = topotest.run_and_expect(test_func, None, count=10, wait=1)
- assertmsg = (
- "BGP session on r1 did not establish connections with one ore more ExaBGP peers"
- )
- assert res is None, assertmsg
-
- # end test_bgp_dampening_setup
-
-
-def test_bgp_dampening_route_announce():
- "Test of BGP route-flap dampening route announcement"
-
- # This test checks if the four ExaBGP peers can announce routes to router
- # r1 and if these routes get forwarded to router r2.
-
- tgen = get_topogen()
-
- # Skip if previous fatal error condition is raised
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info("Starting test of BGP route-flap dampening route announcement")
-
- # Announce routes on exabgp peers to r2
- logger.info("Announcing routes on ExaBGP peers to r1")
- for prefix_iter in range(1, 5):
- for peer_num in range(1, 5):
- pipe = open("/run/exabgp_peer{}.in".format(peer_num), "w")
- with pipe:
- pipe.write(
- "announce route 192.168.{}{}.0/24 next-hop 192.168.101.{}\n".format(
- (peer_num + 2), prefix_iter, (peer_num + 2)
- )
- )
- pipe.close()
- sleep(0.1) # ExaBGP API command processing delay
-
- # Check if routes announced by ExaBGP peers are present in RIB of router r1
- logger.info(
- "Checking if routes announced by ExaBGP peers are present in RIB of router r1"
- )
- router = tgen.gears["r1"]
- reffile = os.path.join(CWD, "r1/bgp_damp_announced.json")
- expected = json.loads(open(reffile).read())
- test_func = functools.partial(
- topotest.router_json_cmp, router, "show ip bgp json", expected
- )
- _, res = topotest.run_and_expect(test_func, None, count=10, wait=1)
- assertmsg = (
- "BGP session on router r1 did not receive routes announced by ExaBGP peers"
- )
- assert res is None, assertmsg
-
- # Check if routes announced by ExaBGP peers to router r1 have been forwarded
- # and are now present in RIB of router r2
- logger.info(
- "Checking if forwarded routes announced by ExaBGP peers are present in RIB of router r2"
- )
- router = tgen.gears["r2"]
- reffile = os.path.join(CWD, "r2/bgp_damp_announced.json")
- expected = json.loads(open(reffile).read())
- test_func = functools.partial(
- topotest.router_json_cmp, router, "show ip bgp json", expected
- )
- _, res = topotest.run_and_expect(test_func, None, count=10, wait=1)
- assertmsg = "BGP session on router r2 did not receive routes announced by ExaBGP peers forwarded by router r1"
- assert res is None, assertmsg
-
- # end test_bgp_dampening_route_announce
-
-
-def test_bgp_dampening_disabled():
- "Test of BGP route-flapping with dampening disabled"
-
- # This test verifies that flapped routes do not get withdrawn from the RIB
- # of router r1 if dampening is disabled.
-
- tgen = get_topogen()
-
- # Skip if previous fatal error condition is raised
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info("Starting test of BGP route-flapping with dampening disabled")
-
- # Flapping routes on ExaBGP peer peer1
- logger.info(
- "Flapping routes on ExaBGP peer peer1 with route-flap dampening disabled"
- )
- for _ in range(1, 5):
- for prefix_iter in range(1, 5):
- pipe = open("/run/exabgp_peer1.in", "w")
- with pipe:
- pipe.write(
- "withdraw route 192.168.3{}.0/24 next-hop 192.168.101.3\n".format(
- prefix_iter
- )
- )
- pipe.close()
- sleep(0.1) # ExaBGP API command processing delay
- sleep(1) # Give the BGP session on router r1 time to process routes
- for prefix_iter in range(1, 5):
- pipe = open("/run/exabgp_peer1.in", "w")
- with pipe:
- pipe.write(
- "announce route 192.168.3{}.0/24 next-hop 192.168.101.3\n".format(
- prefix_iter
- )
- )
- pipe.close()
- sleep(0.1) # ExaBGP API command processing delay
-
- # Verify flapped routes are still present in RIB of router r1
- logger.info(
- "Verifying that the flapped routes are still present in RIB of router r1"
- )
- router = tgen.gears["r1"]
- reffile = os.path.join(CWD, "r1/bgp_damp_announced.json")
- expected = json.loads(open(reffile).read())
- test_func = functools.partial(
- topotest.router_json_cmp, router, "show ip bgp json", expected
- )
- _, res = topotest.run_and_expect(test_func, None, count=10, wait=1)
- assertmsg = "BGP session on router r1 removed flapped routes despite route-flap dampening being disabled"
- assert res is None, assertmsg
-
- # end test_bgp_dampening_disabled
-
-
-def test_bgp_dampening_config():
- "Test of BGP route-flap dampening configuration"
-
- # This test adds peer-group group1 with peers peer1 and peer2 to the
- # configuration of router r1, sets up dampening configurations with
- # different profiles and verifies the configured dampening parameters.
-
- tgen = get_topogen()
- r_1 = tgen.net["r1"]
-
- # Skip if previous fatal error condition is raised
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info("Starting test of BGP route-flap dampening configuration")
-
- # Add peer-group group1 with peers peer1 and peer2
- logger.info(
- "Creating peer-group group1 and adding ExaBGP peers peer1 and peer2 to it"
- )
- r_1.cmd('vtysh -c "conf t" -c "router bgp 65000" -c "neighbor group1 peer-group"')
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "neighbor 192.168.101.3 peer-group group1"'
- ) # peer1
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "neighbor 192.168.101.4 peer-group group1"'
- ) # peer2
-
- # Enable different dampening profiles for peer1, peer3, group1 and global
- # configuration
- logger.info(
- "Enabling different dampening profiles for peer1, peer3, group1 and global configuration"
- )
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "bgp dampening 30 300 900 90"'
- )
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "neighbor group1 dampening 20 200 600 60"'
- )
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "neighbor 192.168.101.3 dampening 10 100 300 30"'
- ) # peer1
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "neighbor 192.168.101.5 dampening 10 100 300 30"'
- ) # peer3
-
- # Verify route-flap dampening configuration
- logger.info("Verifying route-flap dampening configuration on router r1")
- vtyout = r_1.cmd('vtysh -c "show running-config"')
- assertmsg = "BGP Session on r1 does not show enabled global route-flap dampening in running configuration"
- assert re.search("bgp dampening 30 300 900 90", vtyout), assertmsg
- assertmsg = "BGP Session on r1 does not show route-flap dampening enabled for peer-group group1 in running configuration"
- assert re.search("neighbor group1 dampening 20 200 600 60", vtyout), assertmsg
- assertmsg = "BGP Session on r1 does not show route-flap dampening enabled for peer peer1 in running configuration"
- assert re.search(
- "neighbor 192.168.101.3 dampening 10 100 300 30", vtyout
- ), assertmsg
- assertmsg = "BGP Session on r1 does not show route-flap dampening enabled for peer peer3 in running configuration"
- assert re.search(
- "neighbor 192.168.101.5 dampening 10 100 300 30", vtyout
- ), assertmsg
-
- # end test_bgp_dampening_config
-
-
-def test_bgp_dampening_profile_peer_over_group():
- "Test of BGP route-flap dampening profile preferences: peer over group"
-
- # This test verifies that the dampening profile of a peer takes precedence
- # over the dampening profile of its peer-group by flapping the peers routes
- # until dampened and comparing the reuse times to the one specified in the
- # dampening configuration.
-
- tgen = get_topogen()
- r_1 = tgen.net["r1"]
-
- # Skip if previous fatal error condition is raised
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info(
- "Starting test of BGP route-flap dampening profile preferences: peer over group"
- )
-
- # Flapping routes on ExaBGP peer peer1
- logger.info(
- "Flapping routes on ExaBGP peer peer1 with route-flap dampening enabled"
- )
- for _ in range(1, 5):
- for prefix_iter in range(1, 5):
- pipe = open("/run/exabgp_peer1.in", "w")
- with pipe:
- pipe.write(
- "withdraw route 192.168.3{}.0/24 next-hop 192.168.101.3\n".format(
- prefix_iter
- )
- )
- pipe.close()
- sleep(0.1) # ExaBGP API command processing delay
- sleep(1) # Give the BGP session on router r1 time to process routes
- for prefix_iter in range(1, 5):
- pipe = open("/run/exabgp_peer1.in", "w")
- with pipe:
- pipe.write(
- "announce route 192.168.3{}.0/24 next-hop 192.168.101.3\n".format(
- prefix_iter
- )
- )
- pipe.close()
- sleep(0.1) # ExaBGP API command processing delay
-
- # Check damped paths on r1 for routes of peer1 witn peer profile
- logger.info(
- "Checking if router r1 used the correct dampening profile on routes flapped by ExaBGP peer peer1"
- )
- sleep(5) # Wait 5 seconds for paths to show up in dampened-paths list
- vtyout = r_1.cmd('vtysh -c "show ip bgp dampening dampened-paths"')
- routes = re.findall(r"\*d 192\.168\.3\d\.0\/24.*", vtyout)
- assertmsg = (
- "BGP session on router r1 did not dampen routes flapped by ExaBGP peer peer1"
- )
- assert len(routes) == 4, assertmsg
- assertmsg = "BGP session on router r1 used wrong dampening profile for a route flapped by ExaBGP peer peer1"
- for route in routes:
- assert (int(route.split()[3].split(":")[0]) == 0) and ( # hours of reuse time
- 35 > int(route.split()[3].split(":")[1]) > 25
- ), assertmsg # minutes of reuse time
-
- # end test_bgp_dampening_profile_peer_over_group
-
-
-def test_bgp_dampening_profile_group_over_global():
- "Test of BGP route-flap dampening profile preferences: group over global"
-
- # This test verifies that the dampening profile of a peer-group takes
- # precedence over the global dampening profile by flapping the routes of a
- # peer-group member until dampened and comparing the reuse times to the one
- # specified in the dampening configuration.
-
- tgen = get_topogen()
- r_1 = tgen.net["r1"]
-
- # Skip if previous fatal error condition is raised
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info(
- "Starting test of BGP route-flap dampening profile preferences: group over global"
- )
-
- # Flapping routes on ExaBGP peer peer2
- logger.info(
- "Flapping routes on ExaBGP peer peer2 with route-flap dampening enabled"
- )
- for _ in range(1, 5):
- for prefix_iter in range(1, 5):
- pipe = open("/run/exabgp_peer2.in", "w")
- with pipe:
- pipe.write(
- "withdraw route 192.168.4{}.0/24 next-hop 192.168.101.4\n".format(
- prefix_iter
- )
- )
- pipe.close()
- sleep(0.1) # ExaBGP API command processing delay
- sleep(1) # Give the BGP session on router r1 time to process routes
- for prefix_iter in range(1, 5):
- pipe = open("/run/exabgp_peer2.in", "w")
- with pipe:
- pipe.write(
- "announce route 192.168.4{}.0/24 next-hop 192.168.101.4\n".format(
- prefix_iter
- )
- )
- pipe.close()
- sleep(0.1) # ExaBGP API command processing delay
-
- # Check damped paths on r1 for routes of peer2 witn group profile
- logger.info(
- "Checking if router r1 used the correct dampening profile on routes flapped by ExaBGP peer peer2"
- )
- sleep(5) # wait 5 seconds for paths to shop up in damp list
- vtyout = r_1.cmd('vtysh -c "show ip bgp dampening dampened-paths"')
- routes = re.findall(r"\*d 192\.168\.4\d\.0\/24.*", vtyout)
- assertmsg = (
- "BGP session on router r1 did not dampen routes flapped by ExaBGP peer peer2"
- )
- assert len(routes) == 4, assertmsg
- assertmsg = "BGP session on router r1 used wrong dampening profile for a route flapped by ExaBGP peer peer2"
- for route in routes:
- assert (int(route.split()[3].split(":")[0]) == 0) and ( # hours of reuse time
- 65 > int(route.split()[3].split(":")[1]) > 55
- ), assertmsg # minutes of reuse time
-
- # end test_bgp_dampening_profile_group_over_global
-
-
-def test_bgp_dampening_profile_peer_over_global():
- "Test of BGP route-flap dampening profile preferences: peer over global"
-
- # This test verifies that the dampening profile of a peer takes precedence
- # over the global dampening profile by flapping the routes of the peer until
- # dampened and comparing the reuse times to the one specified in the
- # dampening configuration.
-
- tgen = get_topogen()
- r_1 = tgen.net["r1"]
-
- # Skip if previous fatal error condition is raised
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info(
- "Starting test of BGP route-flap dampening profile preferences: peer over global"
- )
-
- # Flapping routes on ExaBGP peer peer3
- logger.info(
- "Flapping routes on ExaBGP peer peer3 with route-flap dampening enabled"
- )
- for _ in range(1, 5):
- for prefix_iter in range(1, 5):
- pipe = open("/run/exabgp_peer3.in", "w")
- with pipe:
- pipe.write(
- "withdraw route 192.168.5{}.0/24 next-hop 192.168.101.5\n".format(
- prefix_iter
- )
- )
- pipe.close()
- sleep(0.1) # ExaBGP API command processing delay
- sleep(1) # Give the BGP session on router r1 time to process routes
- for prefix_iter in range(1, 5):
- pipe = open("/run/exabgp_peer3.in", "w")
- with pipe:
- pipe.write(
- "announce route 192.168.5{}.0/24 next-hop 192.168.101.5\n".format(
- prefix_iter
- )
- )
- pipe.close()
- sleep(0.1) # ExaBGP API command processing delay
-
- # Check damped paths on r1 for routes of peer3 witn peer profile
- logger.info(
- "Checking if router r1 used the correct dampening profile on routes flapped by ExaBGP peer peer3"
- )
- sleep(5) # wait 5 seconds for paths to shop up in damp list
- vtyout = r_1.cmd('vtysh -c "show ip bgp dampening dampened-paths"')
- routes = re.findall(r"\*d 192\.168\.5\d\.0\/24.*", vtyout)
- assertmsg = (
- "BGP session on router r1 did not dampen routes flapped by ExaBGP peer peer3"
- )
- assert len(routes) == 4, assertmsg
- assertmsg = "BGP session on router r1 used wrong dampening profile for a route flapped by ExaBGP peer peer3"
- for route in routes:
- assert (int(route.split()[3].split(":")[0]) == 0) and ( # hours of reuse time
- 35 > int(route.split()[3].split(":")[1]) > 25
- ), assertmsg # minutes of reuse time
-
- # end test_bgp_dampening_profile_peer_over_global
-
-
-def test_bgp_dampening_profile_global():
- "Test of BGP route-flap dampening global profile"
-
- # This test verifies the application of the global dampening profile by
- # flapping the routes of a peer until dampened and comparing the reuse times
- # to the one specified in the dampening configuration.
-
- tgen = get_topogen()
- r_1 = tgen.net["r1"]
-
- # Skip if previous fatal error condition is raised
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info("Starting test of BGP route-flap dampening global profile")
-
- # Flapping routes on ExaBGP peer peer4
- logger.info(
- "Flapping routes on ExaBGP peer peer4 with route-flap dampening enabled"
- )
- for _ in range(1, 5):
- for prefix_iter in range(1, 5):
- pipe = open("/run/exabgp_peer4.in", "w")
- with pipe:
- pipe.write(
- "withdraw route 192.168.6{}.0/24 next-hop 192.168.101.6\n".format(
- prefix_iter
- )
- )
- pipe.close()
- sleep(0.1) # ExaBGP API command processing delay
- sleep(1) # Give the BGP session on router r1 time to process routes
- for prefix_iter in range(1, 5):
- pipe = open("/run/exabgp_peer4.in", "w")
- with pipe:
- pipe.write(
- "announce route 192.168.6{}.0/24 next-hop 192.168.101.6\n".format(
- prefix_iter
- )
- )
- pipe.close()
- sleep(0.1) # ExaBGP API command processing delay
-
- # Check damped paths on r1 for routes of peer4 witn global profile
- logger.info(
- "Checking if router r1 used the global dampening profile on routes flapped by ExaBGP peer peer4"
- )
- sleep(5) # wait 5 seconds for paths to shop up in damp list
- vtyout = r_1.cmd('vtysh -c "show ip bgp dampening dampened-paths"')
- routes = re.findall(r"\*d 192\.168\.6\d\.0\/24.*", vtyout)
- assertmsg = (
- "BGP session on router r1 did not dampen routes flapped by ExaBGP peer peer4"
- )
- assert len(routes) == 4, assertmsg
- assertmsg = "BGP session on router r1 did not use the global dampening profile for a route flapped by ExaBGP peer peer4"
- for route in routes:
- assert (int(route.split()[3].split(":")[0]) == 1) and ( # hours of reuse time
- 35 > int(route.split()[3].split(":")[1]) > 25
- ), assertmsg # minutes of reuse time
-
- # end test_bgp_dampening_profile_global
-
-
-def test_bgp_dampening_withdaw():
- "Test BGP route-flap dampening route withdraw"
-
- # This test verifies that the withrawl of dampened routes from the RIB of
- # router r1 was propagated to router r2.
-
- tgen = get_topogen()
-
- # Skip if previous fatal error condition is raised
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info("Starting test of BGP route-flap dampening route withdraw")
-
- # Check if routes dampened on router r1 have been withdrawn from the RIB on
- # router r2
- logger.info(
- "Checking if routes dampened on router r1 have been withdrawn of RIB on router r2"
- )
- reffile = os.path.join(CWD, "r2/bgp_damp_withdrawn.json")
- expected = json.loads(open(reffile).read())
- test_func = functools.partial(
- topotest.router_json_cmp, tgen.gears["r2"], "show ip bgp json", expected
- )
- _, res = topotest.run_and_expect(test_func, None, count=5, wait=1)
- assertmsg = "BGP session on router r2 did not receive withdraw of routes dampened on router r1"
- assert res is None, assertmsg
-
- # end test_bgp_dampening_withdaw
-
-
-def test_bgp_dampening_cleanup():
- "BGP route-flap dampening test cleanup"
-
- # This test cleans up after other tests associated with route-flap dampening
- # by disabling all dampening configurations, removing added peers and
- # peer-groups from the configuration on router r1, and shutting down ExaBGP
- # peers peer1, peer2 and peer3.
-
- tgen = get_topogen()
- r_1 = tgen.net["r1"]
-
- # Skip if previous fatal error condition is raised
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- logger.info("Starting BGP route-flap dampening test cleanup")
-
- # Disable all dampening configurations
- logger.info("Disabling all dampening configurations")
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "no bgp dampening"'
- )
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "no neighbor group1 dampening"'
- )
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "no neighbor 192.168.101.3 dampening"'
- )
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "address-family ipv4 unicast" -c "no neighbor 192.168.101.5 dampening"'
- )
-
- # Remove ExaBGP peers from configuration of router r1
- logger.info("Removing ExaBGP peers from configuration of router r1")
- for router_num in range(3, 7):
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "no neighbor 192.168.101.{}"'.format(
- router_num
- )
- )
-
- # Remove peer-group group1 from configuration of router r1
- logger.info("Removing peer-group group1 peers from configuration of router r1")
- r_1.cmd(
- 'vtysh -c "conf t" -c "router bgp 65000" -c "no neighbor group1 peer-group"'
- )
-
- # Stop ExaBGP peers and remove associated named pipes
- logger.info("Stopping ExaBGP peers and removing associated named pipes")
- for peer_num in range(1, 5):
- logger.info("Terminating ExaBGP on peer peer{}".format(peer_num))
- peer = tgen.gears["peer{}".format(peer_num)]
- logger.info("Removing named pipe of ExaBGP peer peer{}".format(peer_num))
- fifo_in = "/var/run/exabgp_peer{}.in".format(peer_num)
- peer.stop()
- if os.path.exists(fifo_in):
- os.remove(fifo_in)
-
- # end test_bgp_dampening_cleanup
-
-
-def test_bgp_dampening_aftermath():
- "BGP route-flap dampening aftermath test"
-
- # This test verifies routers r1 and r2 not being affected by the route-flap
- # dampening test series.
-
- tgen = get_topogen()
-
- # Skip if previous fatal error condition is raised
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- # Check BGP Summary on routers r1 and r2
- for rtr_num in [1, 2]:
- logger.info(
- "Checking if BGP router on r{} remains unaffected by route-flap dampening tests".format(
- rtr_num
- )
- )
- router = tgen.gears["r{}".format(rtr_num)]
- reffile = os.path.join(CWD, "r{}/show_bgp.json".format(rtr_num))
- expected = json.loads(open(reffile).read())
- test_func = functools.partial(
- topotest.router_json_cmp, router, "show ip bgp json", expected
- )
- _, res = topotest.run_and_expect(test_func, None, count=10, wait=2)
- assertmsg = "BGP routes on router r{} are wrong after route-flap dampening tests".format(
- rtr_num
- )
- assert res is None, assertmsg
-
- # end test_bgp_dampening_aftermath
-
-
if __name__ == "__main__":
args = ["-s"] + sys.argv[1:]
sys.exit(pytest.main(args))
projects / mirror / frr.git / commitdiff