5 from random import randint
7 from framework import *
13 class BFDAPITestCase(VppTestCase):
14 """Bidirectional Forwarding Detection (BFD) - API"""
18 super(BFDAPITestCase, cls).setUpClass()
21 cls.create_pg_interfaces(range(2))
22 for i in cls.pg_interfaces:
27 super(BFDAPITestCase, cls).tearDownClass()
30 def test_add_bfd(self):
31 """ create a BFD session """
32 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
33 session.add_vpp_config()
34 self.logger.debug("Session state is %s" % str(session.state))
35 session.remove_vpp_config()
36 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
37 session.add_vpp_config()
38 self.logger.debug("Session state is %s" % str(session.state))
39 session.remove_vpp_config()
41 def test_double_add(self):
42 """ create the same BFD session twice (negative case) """
43 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
44 session.add_vpp_config()
46 session.add_vpp_config()
48 session.remove_vpp_config()
50 session.remove_vpp_config()
51 raise Exception("Expected failure while adding duplicate "
54 def test_add_two(self):
55 """ create two BFD sessions """
56 session1 = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
57 session1.add_vpp_config()
58 session2 = VppBFDUDPSession(self, self.pg1, self.pg1.remote_ip4)
59 session2.add_vpp_config()
60 self.assertNotEqual(session1.bs_index, session2.bs_index,
61 "Different BFD sessions share bs_index (%s)" %
65 class BFDTestSession(object):
66 """ BFD session as seen from test framework side """
68 def __init__(self, test, interface, af, detect_mult=3):
71 self.interface = interface
72 self.udp_sport = 50000
74 'my_discriminator': 0,
75 'desired_min_tx_interval': 100000,
76 'detect_mult': detect_mult,
77 'diag': BFDDiagCode.no_diagnostic,
80 def update(self, **kwargs):
81 self.bfd_values.update(kwargs)
83 def create_packet(self):
84 if self.af == AF_INET6:
85 packet = (Ether(src=self.interface.remote_mac,
86 dst=self.interface.local_mac) /
87 IPv6(src=self.interface.remote_ip6,
88 dst=self.interface.local_ip6,
90 UDP(sport=self.udp_sport, dport=BFD.udp_dport) /
93 packet = (Ether(src=self.interface.remote_mac,
94 dst=self.interface.local_mac) /
95 IP(src=self.interface.remote_ip4,
96 dst=self.interface.local_ip4,
98 UDP(sport=self.udp_sport, dport=BFD.udp_dport) /
100 self.test.logger.debug("BFD: Creating packet")
101 for name, value in self.bfd_values.iteritems():
102 self.test.logger.debug("BFD: setting packet.%s=%s", name, value)
103 packet[BFD].setfieldval(name, value)
106 def send_packet(self):
107 p = self.create_packet()
108 self.test.logger.debug(ppp("Sending packet:", p))
109 self.test.pg0.add_stream([p])
112 def verify_packet(self, packet):
113 """ Verify correctness of BFD layer. """
115 self.test.assert_equal(bfd.version, 1, "BFD version")
116 self.test.assert_equal(bfd.your_discriminator,
117 self.bfd_values['my_discriminator'],
118 "BFD - your discriminator")
122 """Common code used by both IPv4 and IPv6 Test Cases"""
125 self.vapi.collect_events() # clear the event queue
126 if not self.vpp_dead:
127 self.vapi.want_bfd_events(enable_disable=0)
128 self.vpp_session.remove_vpp_config()
130 def bfd_session_up(self):
131 self.pg_enable_capture([self.pg0])
132 self.logger.info("BFD: Waiting for slow hello")
133 p, timeout = self.wait_for_bfd_packet(2)
134 self.logger.info("BFD: Sending Init")
135 self.test_session.update(my_discriminator=randint(0, 40000000),
136 your_discriminator=p[BFD].my_discriminator,
138 required_min_rx_interval=100000)
139 self.test_session.send_packet()
140 self.logger.info("BFD: Waiting for event")
141 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
142 self.verify_event(e, expected_state=BFDState.up)
143 self.logger.info("BFD: Session is Up")
144 self.test_session.update(state=BFDState.up)
146 def verify_ip(self, packet):
147 """ Verify correctness of IP layer. """
148 if self.vpp_session.af == AF_INET6:
150 local_ip = self.pg0.local_ip6
151 remote_ip = self.pg0.remote_ip6
152 self.assert_equal(ip.hlim, 255, "IPv6 hop limit")
155 local_ip = self.pg0.local_ip4
156 remote_ip = self.pg0.remote_ip4
157 self.assert_equal(ip.ttl, 255, "IPv4 TTL")
158 self.assert_equal(ip.src, local_ip, "IP source address")
159 self.assert_equal(ip.dst, remote_ip, "IP destination address")
161 def verify_udp(self, packet):
162 """ Verify correctness of UDP layer. """
164 self.assert_equal(udp.dport, BFD.udp_dport, "UDP destination port")
165 self.assert_in_range(udp.sport, BFD.udp_sport_min, BFD.udp_sport_max,
168 def verify_event(self, event, expected_state):
169 """ Verify correctness of event values. """
171 self.logger.debug("BFD: Event: %s" % repr(e))
172 self.assert_equal(e.bs_index, self.vpp_session.bs_index,
176 self.vpp_session.interface.sw_if_index,
177 "BFD interface index")
179 if self.vpp_session.af == AF_INET6:
181 self.assert_equal(e.is_ipv6, is_ipv6, "is_ipv6")
182 if self.vpp_session.af == AF_INET:
183 self.assert_equal(e.local_addr[:4], self.vpp_session.local_addr_n,
184 "Local IPv4 address")
185 self.assert_equal(e.peer_addr[:4], self.vpp_session.peer_addr_n,
188 self.assert_equal(e.local_addr, self.vpp_session.local_addr_n,
189 "Local IPv6 address")
190 self.assert_equal(e.peer_addr, self.vpp_session.peer_addr_n,
192 self.assert_equal(e.state, expected_state, BFDState)
194 def wait_for_bfd_packet(self, timeout=1):
195 """ wait for BFD packet
197 :param timeout: how long to wait max
199 :returns: tuple (packet, time spent waiting for packet)
201 self.logger.info("BFD: Waiting for BFD packet")
203 p = self.pg0.wait_for_packet(timeout=timeout)
205 self.logger.debug(ppp("Got packet:", p))
208 raise Exception(ppp("Unexpected or invalid BFD packet:", p))
210 raise Exception(ppp("Unexpected payload in BFD packet:", bfd))
213 self.test_session.verify_packet(p)
214 return p, after - before
216 def test_session_up(self):
217 """ bring BFD session up """
218 self.bfd_session_up()
220 def test_hold_up(self):
221 """ hold BFD session up """
222 self.bfd_session_up()
224 self.wait_for_bfd_packet()
225 self.test_session.send_packet()
228 class BFD4TestCase(VppTestCase, BFDCommonCode):
229 """Bidirectional Forwarding Detection (BFD)"""
233 super(BFD4TestCase, cls).setUpClass()
235 cls.create_pg_interfaces([0])
237 cls.pg0.generate_remote_hosts()
238 cls.pg0.configure_ipv4_neighbors()
240 cls.pg0.resolve_arp()
243 super(BFD4TestCase, cls).tearDownClass()
247 super(BFD4TestCase, self).setUp()
248 self.vapi.want_bfd_events()
250 self.vpp_session = VppBFDUDPSession(self, self.pg0,
252 self.vpp_session.add_vpp_config()
253 self.vpp_session.admin_up()
254 self.test_session = BFDTestSession(self, self.pg0, AF_INET)
256 self.vapi.want_bfd_events(enable_disable=0)
260 BFDCommonCode.tearDown(self)
261 super(BFD4TestCase, self).tearDown()
263 def test_slow_timer(self):
264 """ verify slow periodic control frames while session down """
265 self.pg_enable_capture([self.pg0])
267 self.logger.info("BFD: Waiting for %d BFD packets" % expected_packets)
268 self.wait_for_bfd_packet(2)
269 for i in range(expected_packets):
271 self.wait_for_bfd_packet(2)
273 # spec says the range should be <0.75, 1>, allow extra 0.05 margin
274 # to work around timing issues
275 self.assert_in_range(
276 after - before, 0.70, 1.05, "time between slow packets")
279 def test_zero_remote_min_rx(self):
280 """ no packets when zero BFD RemoteMinRxInterval """
281 self.pg_enable_capture([self.pg0])
282 p, timeout = self.wait_for_bfd_packet(2)
283 self.test_session.update(my_discriminator=randint(0, 40000000),
284 your_discriminator=p[BFD].my_discriminator,
286 required_min_rx_interval=0)
287 self.test_session.send_packet()
288 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
289 self.verify_event(e, expected_state=BFDState.up)
292 p = self.pg0.wait_for_packet(timeout=1)
295 raise Exception(ppp("Received unexpected BFD packet:", p))
297 def test_conn_down(self):
298 """ verify session goes down after inactivity """
299 self.bfd_session_up()
300 self.wait_for_bfd_packet()
301 self.assert_equal(len(self.vapi.collect_events()), 0,
302 "number of bfd events")
303 self.wait_for_bfd_packet()
304 self.assert_equal(len(self.vapi.collect_events()), 0,
305 "number of bfd events")
306 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
307 self.verify_event(e, expected_state=BFDState.down)
309 def test_large_required_min_rx(self):
310 """ large remote RequiredMinRxInterval """
311 self.bfd_session_up()
313 self.test_session.update(required_min_rx_interval=interval)
314 self.test_session.send_packet()
317 while time.time() < now + interval / us_in_sec:
319 p = self.wait_for_bfd_packet()
321 self.logger.error(ppp("Received unexpected packet:", p))
325 self.assert_in_range(count, 0, 1, "number of packets received")
327 def test_immediate_remote_min_rx_reduce(self):
328 """ immediately honor remote min rx reduction """
329 self.vpp_session.remove_vpp_config()
330 self.vpp_session = VppBFDUDPSession(
331 self, self.pg0, self.pg0.remote_ip4, desired_min_tx=10000)
332 self.vpp_session.add_vpp_config()
333 self.test_session.update(desired_min_tx_interval=1000000,
334 required_min_rx_interval=1000000)
335 self.bfd_session_up()
336 self.wait_for_bfd_packet()
338 self.test_session.update(required_min_rx_interval=interval)
339 self.test_session.send_packet()
340 p, ttp = self.wait_for_bfd_packet()
341 # allow extra 10% to work around timing issues, first packet is special
342 self.assert_in_range(ttp, 0, 1.10 * interval / us_in_sec,
343 "time between BFD packets")
344 p, ttp = self.wait_for_bfd_packet()
345 self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec,
346 1.10 * interval / us_in_sec,
347 "time between BFD packets")
348 p, ttp = self.wait_for_bfd_packet()
349 self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec,
350 1.10 * interval / us_in_sec,
351 "time between BFD packets")
354 class BFD6TestCase(VppTestCase, BFDCommonCode):
355 """Bidirectional Forwarding Detection (BFD) (IPv6) """
359 super(BFD6TestCase, cls).setUpClass()
361 cls.create_pg_interfaces([0])
363 cls.pg0.configure_ipv6_neighbors()
365 cls.pg0.resolve_ndp()
368 super(BFD6TestCase, cls).tearDownClass()
372 super(BFD6TestCase, self).setUp()
373 self.vapi.want_bfd_events()
375 self.vpp_session = VppBFDUDPSession(self, self.pg0,
378 self.vpp_session.add_vpp_config()
379 self.vpp_session.admin_up()
380 self.test_session = BFDTestSession(self, self.pg0, AF_INET6)
381 self.logger.debug(self.vapi.cli("show adj nbr"))
383 self.vapi.want_bfd_events(enable_disable=0)
387 BFDCommonCode.tearDown(self)
388 super(BFD6TestCase, self).tearDown()
390 if __name__ == '__main__':
391 unittest.main(testRunner=VppTestRunner)