2 # SPDX-License-Identifier: ISC
5 # test_bgp_update_delay.py
6 # Part of NetDEF Topology Tests
8 # Copyright (c) 2019 by
9 # Don Slice <dslice@nvidia.com>
13 Test the ability to define update-delay to delay bestpath, rib install
14 and advertisement to peers when frr is started, restarted or "clear ip
15 bgp *" is performed. Test both the vrf-specific and global configuration
26 r2 is UUT and peers with r1, r3, and r4 in default bgp instance.
27 r2 peers with r5 in vrf vrf1.
29 Check r2 initial convergence in default table
30 Define update-delay with max-delay in the default bgp instance on r2
31 Shutdown peering on r1 toward r2 so that delay timers can be exercised
32 Clear bgp neighbors on r2 and then check for the 'in progress' indicator
33 Check that r2 only installs route learned from r4 after the max-delay timer expires
34 Define update-delay with max-delay and estabish-wait and check json output showing set
35 Clear neighbors on r2 and check that r3 installs route from r4 after establish-wait time
36 Remove update-delay timer on r2 to verify that it goes back to normal behavior
37 Clear neighbors on r2 and check that route install time on r2 does not delay
38 Define global bgp update-delay with max-delay and establish-wait on r2
39 Check that r2 default instance and vrf1 have the max-delay and establish set
40 Clear neighbors on r2 and check route-install time is after the establish-wait timer
42 Note that the keepalive/hold times were changed to 3/9 and the connect retry timer
43 to 10 to improve the odds the convergence timing in this test case is useful in the
53 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
54 sys
.path
.append(os
.path
.join(CWD
, "../"))
56 # pylint: disable=C0413
57 from lib
import topotest
58 from lib
.topogen
import Topogen
, TopoRouter
, get_topogen
60 pytestmark
= [pytest
.mark
.bgpd
]
63 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
67 for routern
in range(1, 6):
68 tgen
.add_router("r{}".format(routern
))
70 switch
= tgen
.add_switch("s1")
71 switch
.add_link(tgen
.gears
["r1"])
72 switch
.add_link(tgen
.gears
["r2"])
74 switch
= tgen
.add_switch("s2")
75 switch
.add_link(tgen
.gears
["r2"])
76 switch
.add_link(tgen
.gears
["r3"])
78 switch
= tgen
.add_switch("s3")
79 switch
.add_link(tgen
.gears
["r2"])
80 switch
.add_link(tgen
.gears
["r4"])
82 switch
= tgen
.add_switch("s4")
83 switch
.add_link(tgen
.gears
["r2"])
84 switch
.add_link(tgen
.gears
["r5"])
87 def setup_module(mod
):
88 tgen
= Topogen(build_topo
, mod
.__name
__)
91 router_list
= tgen
.routers()
93 for i
, (rname
, router
) in enumerate(router_list
.items(), 1):
95 TopoRouter
.RD_ZEBRA
, os
.path
.join(CWD
, "{}/zebra.conf".format(rname
))
98 TopoRouter
.RD_BGP
, os
.path
.join(CWD
, "{}/bgpd.conf".format(rname
))
104 def teardown_module(mod
):
109 def test_bgp_update_delay():
112 if tgen
.routers_have_failure():
113 pytest
.skip(tgen
.errors
)
115 router1
= tgen
.gears
["r1"]
116 router2
= tgen
.gears
["r2"]
117 router3
= tgen
.gears
["r3"]
119 # initial convergence without update-delay defined
120 def _bgp_converge(router
):
121 output
= json
.loads(router
.vtysh_cmd("show ip bgp neighbor 192.168.255.2 json"))
124 "bgpState": "Established",
125 "addressFamilyInfo": {"ipv4Unicast": {"acceptedPrefixCounter": 2}},
128 return topotest
.json_cmp(output
, expected
)
130 def _bgp_check_update_delay(router
):
131 output
= json
.loads(router
.vtysh_cmd("show ip bgp sum json"))
132 expected
= {"ipv4Unicast": {"updateDelayLimit": 20}}
134 return topotest
.json_cmp(output
, expected
)
136 def _bgp_check_update_delay_in_progress(router
):
137 output
= json
.loads(router
.vtysh_cmd("show ip bgp sum json"))
138 expected
= {"ipv4Unicast": {"updateDelayInProgress": True}}
140 return topotest
.json_cmp(output
, expected
)
142 def _bgp_check_route_install(router
):
143 output
= json
.loads(router
.vtysh_cmd("show ip route 172.16.253.254/32 json"))
144 expected
= {"172.16.253.254/32": [{"protocol": "bgp"}]}
146 return topotest
.json_cmp(output
, expected
)
148 def _bgp_check_update_delay_and_wait(router
):
149 output
= json
.loads(router
.vtysh_cmd("show ip bgp sum json"))
151 "ipv4Unicast": {"updateDelayLimit": 20, "updateDelayEstablishWait": 10}
154 return topotest
.json_cmp(output
, expected
)
156 def _bgp_check_update_delay(router
):
157 output
= json
.loads(router
.vtysh_cmd("show ip bgp sum json"))
158 expected
= {"ipv4Unicast": {"updateDelayLimit": 20}}
160 return topotest
.json_cmp(output
, expected
)
162 def _bgp_check_vrf_update_delay_and_wait(router
):
163 output
= json
.loads(router
.vtysh_cmd("show ip bgp vrf vrf1 sum json"))
165 "ipv4Unicast": {"updateDelayLimit": 20, "updateDelayEstablishWait": 10}
168 return topotest
.json_cmp(output
, expected
)
170 # Check r2 initial convergence in default table
171 test_func
= functools
.partial(_bgp_converge
, router2
)
172 success
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
174 assert result
is None, 'Failed bgp convergence in "{}"'.format(router2
)
176 # Define update-delay with max-delay in the default bgp instance on r2
185 # Shutdown peering on r1 toward r2 so that delay timers can be exercised
190 neighbor 192.168.255.1 shut
194 # Clear bgp neighbors on r2 and then check for the 'in progress' indicator
195 router2
.vtysh_cmd("""clear ip bgp *""")
197 test_func
= functools
.partial(_bgp_check_update_delay_in_progress
, router2
)
198 success
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
200 assert result
is None, 'Failed to set update-delay max-delay timer "{}"'.format(
204 # Check that r2 only installs route learned from r4 after the max-delay timer expires
205 test_func
= functools
.partial(_bgp_check_route_install
, router2
)
206 success
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
208 assert result
is None, 'Failed to install route after update-delay "{}"'.format(
212 # Define update-delay with max-delay and estabish-wait and check json output showing set
221 test_func
= functools
.partial(_bgp_check_update_delay_and_wait
, router2
)
222 success
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
226 ), 'Failed to set max-delay and establish-weight timers in "{}"'.format(router2
)
228 # Define update-delay with max-delay and estabish-wait and check json output showing set
229 router2
.vtysh_cmd("""clear ip bgp *""")
231 test_func
= functools
.partial(_bgp_check_route_install
, router3
)
232 success
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
236 ), 'Failed to installed advertised route after establish-wait timer espired "{}"'.format(
240 # Remove update-delay timer on r2 to verify that it goes back to normal behavior
249 # Clear neighbors on r2 and check that route install time on r2 does not delay
250 router2
.vtysh_cmd("""clear ip bgp *""")
252 test_func
= functools
.partial(_bgp_check_route_install
, router2
)
253 success
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
255 assert result
is None, 'Failed to remove update-delay delay timing "{}"'.format(
259 # Define global bgp update-delay with max-delay and establish-wait on r2
263 bgp update-delay 20 10
267 # Check that r2 default instance and vrf1 have the max-delay and establish set
268 test_func
= functools
.partial(_bgp_check_update_delay_and_wait
, router2
)
269 success
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
271 assert result
is None, 'Failed to set update-delay in default instance "{}"'.format(
275 test_func
= functools
.partial(_bgp_check_vrf_update_delay_and_wait
, router2
)
276 success
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
278 assert result
is None, 'Failed to set update-delay in vrf1 "{}"'.format(router2
)
280 # Clear neighbors on r2 and check route-install time is after the establish-wait timer
281 router2
.vtysh_cmd("""clear ip bgp *""")
283 test_func
= functools
.partial(_bgp_check_route_install
, router3
)
284 success
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
288 ), 'Failed to installed advertised route after establish-wait timer espired "{}"'.format(
293 if __name__
== "__main__":
294 args
= ["-s"] + sys
.argv
[1:]
295 sys
.exit(pytest
.main(args
))