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 with self.vapi.expect_negative_api_retval():
47 session.add_vpp_config()
49 session.remove_vpp_config()
51 def test_add_two(self):
52 """ create two BFD sessions """
53 session1 = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
54 session1.add_vpp_config()
55 session2 = VppBFDUDPSession(self, self.pg1, self.pg1.remote_ip4)
56 session2.add_vpp_config()
57 self.assertNotEqual(session1.bs_index, session2.bs_index,
58 "Different BFD sessions share bs_index (%s)" %
62 class BFDTestSession(object):
63 """ BFD session as seen from test framework side """
65 def __init__(self, test, interface, af, detect_mult=3):
68 self.interface = interface
69 self.udp_sport = 50000
71 'my_discriminator': 0,
72 'desired_min_tx_interval': 100000,
73 'detect_mult': detect_mult,
74 'diag': BFDDiagCode.no_diagnostic,
77 def update(self, **kwargs):
78 self.bfd_values.update(kwargs)
80 def create_packet(self):
81 if self.af == AF_INET6:
82 packet = (Ether(src=self.interface.remote_mac,
83 dst=self.interface.local_mac) /
84 IPv6(src=self.interface.remote_ip6,
85 dst=self.interface.local_ip6,
87 UDP(sport=self.udp_sport, dport=BFD.udp_dport) /
90 packet = (Ether(src=self.interface.remote_mac,
91 dst=self.interface.local_mac) /
92 IP(src=self.interface.remote_ip4,
93 dst=self.interface.local_ip4,
95 UDP(sport=self.udp_sport, dport=BFD.udp_dport) /
97 self.test.logger.debug("BFD: Creating packet")
98 for name, value in self.bfd_values.iteritems():
99 self.test.logger.debug("BFD: setting packet.%s=%s", name, value)
100 packet[BFD].setfieldval(name, value)
103 def send_packet(self):
104 p = self.create_packet()
105 self.test.logger.debug(ppp("Sending packet:", p))
106 self.test.pg0.add_stream([p])
109 def verify_packet(self, packet):
110 """ Verify correctness of BFD layer. """
112 self.test.assert_equal(bfd.version, 1, "BFD version")
113 self.test.assert_equal(bfd.your_discriminator,
114 self.bfd_values['my_discriminator'],
115 "BFD - your discriminator")
119 """Common code used by both IPv4 and IPv6 Test Cases"""
122 self.vapi.collect_events() # clear the event queue
123 if not self.vpp_dead:
124 self.vapi.want_bfd_events(enable_disable=0)
125 self.vpp_session.remove_vpp_config()
127 def bfd_session_up(self):
128 self.pg_enable_capture([self.pg0])
129 self.logger.info("BFD: Waiting for slow hello")
130 p, timeout = self.wait_for_bfd_packet(2)
131 self.logger.info("BFD: Sending Init")
132 self.test_session.update(my_discriminator=randint(0, 40000000),
133 your_discriminator=p[BFD].my_discriminator,
135 required_min_rx_interval=100000)
136 self.test_session.send_packet()
137 self.logger.info("BFD: Waiting for event")
138 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
139 self.verify_event(e, expected_state=BFDState.up)
140 self.logger.info("BFD: Session is Up")
141 self.test_session.update(state=BFDState.up)
143 def verify_ip(self, packet):
144 """ Verify correctness of IP layer. """
145 if self.vpp_session.af == AF_INET6:
147 local_ip = self.pg0.local_ip6
148 remote_ip = self.pg0.remote_ip6
149 self.assert_equal(ip.hlim, 255, "IPv6 hop limit")
152 local_ip = self.pg0.local_ip4
153 remote_ip = self.pg0.remote_ip4
154 self.assert_equal(ip.ttl, 255, "IPv4 TTL")
155 self.assert_equal(ip.src, local_ip, "IP source address")
156 self.assert_equal(ip.dst, remote_ip, "IP destination address")
158 def verify_udp(self, packet):
159 """ Verify correctness of UDP layer. """
161 self.assert_equal(udp.dport, BFD.udp_dport, "UDP destination port")
162 self.assert_in_range(udp.sport, BFD.udp_sport_min, BFD.udp_sport_max,
165 def verify_event(self, event, expected_state):
166 """ Verify correctness of event values. """
168 self.logger.debug("BFD: Event: %s" % repr(e))
169 self.assert_equal(e.bs_index, self.vpp_session.bs_index,
173 self.vpp_session.interface.sw_if_index,
174 "BFD interface index")
176 if self.vpp_session.af == AF_INET6:
178 self.assert_equal(e.is_ipv6, is_ipv6, "is_ipv6")
179 if self.vpp_session.af == AF_INET:
180 self.assert_equal(e.local_addr[:4], self.vpp_session.local_addr_n,
181 "Local IPv4 address")
182 self.assert_equal(e.peer_addr[:4], self.vpp_session.peer_addr_n,
185 self.assert_equal(e.local_addr, self.vpp_session.local_addr_n,
186 "Local IPv6 address")
187 self.assert_equal(e.peer_addr, self.vpp_session.peer_addr_n,
189 self.assert_equal(e.state, expected_state, BFDState)
191 def wait_for_bfd_packet(self, timeout=1):
192 """ wait for BFD packet
194 :param timeout: how long to wait max
196 :returns: tuple (packet, time spent waiting for packet)
198 self.logger.info("BFD: Waiting for BFD packet")
200 p = self.pg0.wait_for_packet(timeout=timeout)
202 self.logger.debug(ppp("Got packet:", p))
205 raise Exception(ppp("Unexpected or invalid BFD packet:", p))
207 raise Exception(ppp("Unexpected payload in BFD packet:", bfd))
210 self.test_session.verify_packet(p)
211 return p, after - before
213 def test_session_up(self):
214 """ bring BFD session up """
215 self.bfd_session_up()
217 def test_hold_up(self):
218 """ hold BFD session up """
219 self.bfd_session_up()
221 self.wait_for_bfd_packet()
222 self.test_session.send_packet()
225 class BFD4TestCase(VppTestCase, BFDCommonCode):
226 """Bidirectional Forwarding Detection (BFD)"""
230 super(BFD4TestCase, cls).setUpClass()
232 cls.create_pg_interfaces([0])
234 cls.pg0.generate_remote_hosts()
235 cls.pg0.configure_ipv4_neighbors()
237 cls.pg0.resolve_arp()
240 super(BFD4TestCase, cls).tearDownClass()
244 super(BFD4TestCase, self).setUp()
245 self.vapi.want_bfd_events()
247 self.vpp_session = VppBFDUDPSession(self, self.pg0,
249 self.vpp_session.add_vpp_config()
250 self.vpp_session.admin_up()
251 self.test_session = BFDTestSession(self, self.pg0, AF_INET)
253 self.vapi.want_bfd_events(enable_disable=0)
257 BFDCommonCode.tearDown(self)
258 super(BFD4TestCase, self).tearDown()
260 def test_slow_timer(self):
261 """ verify slow periodic control frames while session down """
262 self.pg_enable_capture([self.pg0])
264 self.logger.info("BFD: Waiting for %d BFD packets" % expected_packets)
265 self.wait_for_bfd_packet(2)
266 for i in range(expected_packets):
268 self.wait_for_bfd_packet(2)
270 # spec says the range should be <0.75, 1>, allow extra 0.05 margin
271 # to work around timing issues
272 self.assert_in_range(
273 after - before, 0.70, 1.05, "time between slow packets")
276 def test_zero_remote_min_rx(self):
277 """ no packets when zero BFD RemoteMinRxInterval """
278 self.pg_enable_capture([self.pg0])
279 p, timeout = self.wait_for_bfd_packet(2)
280 self.test_session.update(my_discriminator=randint(0, 40000000),
281 your_discriminator=p[BFD].my_discriminator,
283 required_min_rx_interval=0)
284 self.test_session.send_packet()
285 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
286 self.verify_event(e, expected_state=BFDState.up)
289 p = self.pg0.wait_for_packet(timeout=1)
292 raise Exception(ppp("Received unexpected BFD packet:", p))
294 def test_conn_down(self):
295 """ verify session goes down after inactivity """
296 self.bfd_session_up()
297 self.wait_for_bfd_packet()
298 self.assert_equal(len(self.vapi.collect_events()), 0,
299 "number of bfd events")
300 self.wait_for_bfd_packet()
301 self.assert_equal(len(self.vapi.collect_events()), 0,
302 "number of bfd events")
303 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
304 self.verify_event(e, expected_state=BFDState.down)
306 def test_large_required_min_rx(self):
307 """ large remote RequiredMinRxInterval """
308 self.bfd_session_up()
310 self.test_session.update(required_min_rx_interval=interval)
311 self.test_session.send_packet()
314 while time.time() < now + interval / us_in_sec:
316 p = self.wait_for_bfd_packet()
318 self.logger.error(ppp("Received unexpected packet:", p))
322 self.assert_in_range(count, 0, 1, "number of packets received")
324 def test_immediate_remote_min_rx_reduce(self):
325 """ immediately honor remote min rx reduction """
326 self.vpp_session.remove_vpp_config()
327 self.vpp_session = VppBFDUDPSession(
328 self, self.pg0, self.pg0.remote_ip4, desired_min_tx=10000)
329 self.vpp_session.add_vpp_config()
330 self.test_session.update(desired_min_tx_interval=1000000,
331 required_min_rx_interval=1000000)
332 self.bfd_session_up()
333 self.wait_for_bfd_packet()
335 self.test_session.update(required_min_rx_interval=interval)
336 self.test_session.send_packet()
337 p, ttp = self.wait_for_bfd_packet()
338 # allow extra 10% to work around timing issues, first packet is special
339 self.assert_in_range(ttp, 0, 1.10 * interval / us_in_sec,
340 "time between BFD packets")
341 p, ttp = self.wait_for_bfd_packet()
342 self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec,
343 1.10 * interval / us_in_sec,
344 "time between BFD packets")
345 p, ttp = self.wait_for_bfd_packet()
346 self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec,
347 1.10 * interval / us_in_sec,
348 "time between BFD packets")
351 class BFD6TestCase(VppTestCase, BFDCommonCode):
352 """Bidirectional Forwarding Detection (BFD) (IPv6) """
356 super(BFD6TestCase, cls).setUpClass()
358 cls.create_pg_interfaces([0])
360 cls.pg0.configure_ipv6_neighbors()
362 cls.pg0.resolve_ndp()
365 super(BFD6TestCase, cls).tearDownClass()
369 super(BFD6TestCase, self).setUp()
370 self.vapi.want_bfd_events()
372 self.vpp_session = VppBFDUDPSession(self, self.pg0,
375 self.vpp_session.add_vpp_config()
376 self.vpp_session.admin_up()
377 self.test_session = BFDTestSession(self, self.pg0, AF_INET6)
378 self.logger.debug(self.vapi.cli("show adj nbr"))
380 self.vapi.want_bfd_events(enable_disable=0)
384 BFDCommonCode.tearDown(self)
385 super(BFD6TestCase, self).tearDown()
387 if __name__ == '__main__':
388 unittest.main(testRunner=VppTestRunner)