]>
Commit | Line | Data |
---|---|---|
97242660 | 1 | #!/usr/bin/env python |
acddc0ed | 2 | # SPDX-License-Identifier: ISC |
97242660 DA |
3 | |
4 | # | |
5 | # bgp_gr_notification.py | |
6 | # | |
7 | # Copyright (c) 2022 by | |
8 | # Donatas Abraitis <donatas@opensourcerouting.org> | |
9 | # | |
97242660 DA |
10 | |
11 | """ | |
12 | TC1: Disable the link between R1-R2 and wait for HoldTimerExpire notification: | |
13 | 1) Check if R2 sent HoldTimerExpired notification | |
14 | 2) Check if the routes are retained at R2 | |
15 | TC2: Trigger `clear bgp` (Administrative Reset): | |
16 | `bgp hard-administrative-reset` disabled: | |
17 | a) Check if Administrative Reset notification was sent from R2 | |
18 | b) Routes should be retained on R1 | |
19 | """ | |
20 | ||
21 | import os | |
22 | import sys | |
23 | import json | |
24 | import pytest | |
25 | import functools | |
26 | ||
27 | CWD = os.path.dirname(os.path.realpath(__file__)) | |
28 | sys.path.append(os.path.join(CWD, "../")) | |
29 | ||
30 | # pylint: disable=C0413 | |
31 | from lib import topotest | |
32 | from lib.topogen import Topogen, TopoRouter, get_topogen | |
33 | from lib.common_config import step | |
34 | ||
35 | pytestmark = [pytest.mark.bgpd] | |
36 | ||
37 | ||
38 | def build_topo(tgen): | |
39 | for routern in range(1, 3): | |
40 | tgen.add_router("r{}".format(routern)) | |
41 | ||
42 | switch = tgen.add_switch("s1") | |
43 | switch.add_link(tgen.gears["r1"]) | |
44 | switch.add_link(tgen.gears["r2"]) | |
45 | ||
46 | ||
47 | def setup_module(mod): | |
48 | tgen = Topogen(build_topo, mod.__name__) | |
49 | tgen.start_topology() | |
50 | ||
51 | router_list = tgen.routers() | |
52 | ||
53 | for i, (rname, router) in enumerate(router_list.items(), 1): | |
54 | router.load_config( | |
55 | TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname)) | |
56 | ) | |
57 | router.load_config( | |
58 | TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname)) | |
59 | ) | |
60 | ||
61 | tgen.start_router() | |
62 | ||
63 | ||
64 | def teardown_module(mod): | |
65 | tgen = get_topogen() | |
66 | tgen.stop_topology() | |
67 | ||
68 | ||
69 | def test_bgp_hold_timer_expired_gr(): | |
70 | # TC1 | |
71 | tgen = get_topogen() | |
72 | ||
73 | if tgen.routers_have_failure(): | |
74 | pytest.skip(tgen.errors) | |
75 | ||
76 | r1 = tgen.gears["r1"] | |
77 | r2 = tgen.gears["r2"] | |
78 | ||
79 | def _bgp_converge(): | |
80 | output = json.loads(r2.vtysh_cmd("show ip bgp neighbor 192.168.255.1 json")) | |
81 | expected = { | |
82 | "192.168.255.1": { | |
83 | "bgpState": "Established", | |
84 | "addressFamilyInfo": {"ipv4Unicast": {"acceptedPrefixCounter": 2}}, | |
85 | } | |
86 | } | |
87 | return topotest.json_cmp(output, expected) | |
88 | ||
89 | def _disable_link_r1_r2(): | |
90 | r1.cmd_raises("ip link set down dev r1-eth0") | |
91 | ||
92 | def _enable_link_r1_r2(): | |
93 | r1.cmd_raises("ip link set up dev r1-eth0") | |
94 | ||
95 | def _bgp_check_hold_timer_expire_reason(): | |
96 | output = json.loads(r2.vtysh_cmd("show ip bgp neighbor 192.168.255.1 json")) | |
97 | expected = { | |
98 | "192.168.255.1": { | |
99 | "lastNotificationReason": "Hold Timer Expired", | |
100 | } | |
101 | } | |
102 | return topotest.json_cmp(output, expected) | |
103 | ||
104 | def _bgp_check_hold_timer_expire_stale(): | |
105 | output = json.loads(r2.vtysh_cmd("show bgp ipv4 unicast 172.16.255.1/32 json")) | |
106 | expected = { | |
107 | "paths": [ | |
108 | { | |
109 | "stale": True, | |
110 | "valid": True, | |
111 | } | |
112 | ] | |
113 | } | |
114 | return topotest.json_cmp(output, expected) | |
115 | ||
116 | step("Initial BGP converge") | |
117 | test_func = functools.partial(_bgp_converge) | |
118 | _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) | |
119 | assert result is None, "Failed to see BGP convergence on R2" | |
120 | ||
121 | step("Disable the link between R1-R2") | |
122 | _disable_link_r1_r2() | |
123 | ||
124 | step("Check if R2 sent HoldTimerExpire notification to R1") | |
125 | test_func = functools.partial(_bgp_check_hold_timer_expire_reason) | |
126 | _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) | |
127 | assert result is None, "Failed to see Hold Timer Expired notification from R2 on R1" | |
128 | ||
129 | step("Check if the routes are retained at R2") | |
130 | test_func = functools.partial(_bgp_check_hold_timer_expire_stale) | |
131 | _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) | |
132 | assert result is None, "Failed to see retained stale routes on R2" | |
133 | ||
134 | step("Enable the link between R1-R2") | |
135 | _enable_link_r1_r2() | |
136 | ||
137 | ||
138 | def test_bgp_administrative_reset_gr(): | |
139 | # TC2 | |
140 | tgen = get_topogen() | |
141 | ||
142 | if tgen.routers_have_failure(): | |
143 | pytest.skip(tgen.errors) | |
144 | ||
145 | r1 = tgen.gears["r1"] | |
146 | r2 = tgen.gears["r2"] | |
147 | ||
148 | def _bgp_converge(): | |
149 | output = json.loads(r2.vtysh_cmd("show ip bgp neighbor 192.168.255.1 json")) | |
150 | expected = { | |
151 | "192.168.255.1": { | |
152 | "bgpState": "Established", | |
153 | "addressFamilyInfo": {"ipv4Unicast": {"acceptedPrefixCounter": 2}}, | |
154 | } | |
155 | } | |
156 | return topotest.json_cmp(output, expected) | |
157 | ||
158 | def _bgp_check_hard_reset(): | |
159 | output = json.loads(r1.vtysh_cmd("show ip bgp neighbor 192.168.255.2 json")) | |
160 | expected = { | |
161 | "192.168.255.2": { | |
162 | "lastNotificationReason": "Cease/Administrative Reset", | |
163 | "lastNotificationHardReset": False, | |
164 | } | |
165 | } | |
166 | return topotest.json_cmp(output, expected) | |
167 | ||
168 | def _bgp_check_gr_notification_stale(): | |
169 | output = json.loads(r1.vtysh_cmd("show bgp ipv4 unicast 172.16.255.2/32 json")) | |
170 | expected = { | |
171 | "paths": [ | |
172 | { | |
173 | "stale": True, | |
174 | "valid": True, | |
175 | } | |
176 | ] | |
177 | } | |
178 | return topotest.json_cmp(output, expected) | |
179 | ||
180 | def _bgp_clear_r1_and_shutdown(): | |
181 | r2.vtysh_cmd( | |
182 | """ | |
183 | clear ip bgp 192.168.255.1 | |
184 | configure terminal | |
185 | router bgp | |
186 | neighbor 192.168.255.1 shutdown | |
187 | """ | |
188 | ) | |
189 | ||
190 | step("Initial BGP converge") | |
191 | test_func = functools.partial(_bgp_converge) | |
192 | _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) | |
193 | assert result is None, "Failed to see BGP convergence on R2" | |
194 | ||
195 | step("Reset and shutdown R1") | |
196 | _bgp_clear_r1_and_shutdown() | |
197 | ||
198 | step("Check if Hard Reset notification wasn't sent from R2") | |
199 | test_func = functools.partial(_bgp_check_hard_reset) | |
200 | _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) | |
201 | assert result is None, "Failed to send Administrative Reset notification from R2" | |
202 | ||
203 | step("Check if stale routes are retained on R1") | |
204 | test_func = functools.partial(_bgp_check_gr_notification_stale) | |
205 | _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5) | |
206 | assert result is None, "Failed to see retained stale routes on R1" | |
207 | ||
208 | ||
209 | if __name__ == "__main__": | |
210 | args = ["-s"] + sys.argv[1:] | |
211 | sys.exit(pytest.main(args)) |