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()
32 def test_add_bfd(self):
33 """ create a BFD session """
34 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
35 session.add_vpp_config()
36 self.logger.debug("Session state is %s" % str(session.state))
37 session.remove_vpp_config()
38 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
39 session.add_vpp_config()
40 self.logger.debug("Session state is %s" % str(session.state))
41 session.remove_vpp_config()
43 def test_double_add(self):
44 """ create the same BFD session twice (negative case) """
45 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
46 session.add_vpp_config()
48 session.add_vpp_config()
50 session.remove_vpp_config()
52 session.remove_vpp_config()
53 raise Exception("Expected failure while adding duplicate "
56 def test_add_two(self):
57 """ create two BFD sessions """
58 session1 = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
59 session1.add_vpp_config()
60 session2 = VppBFDUDPSession(self, self.pg1, self.pg1.remote_ip4)
61 session2.add_vpp_config()
62 self.assertNotEqual(session1.bs_index, session2.bs_index,
63 "Different BFD sessions share bs_index (%s)" %
67 def create_packet(interface, ttl=255, src_port=50000, **kwargs):
68 p = (Ether(src=interface.remote_mac, dst=interface.local_mac) /
69 IP(src=interface.remote_ip4, dst=interface.local_ip4, ttl=ttl) /
70 UDP(sport=src_port, dport=BFD.udp_dport) /
75 def verify_ip(test, packet, local_ip, remote_ip):
76 """ Verify correctness of IP layer. """
78 test.assert_equal(ip.src, local_ip, "IP source address")
79 test.assert_equal(ip.dst, remote_ip, "IP destination address")
80 test.assert_equal(ip.ttl, 255, "IP TTL")
83 def verify_udp(test, packet):
84 """ Verify correctness of UDP layer. """
86 test.assert_equal(udp.dport, BFD.udp_dport, "UDP destination port")
87 test.assert_in_range(udp.sport, BFD.udp_sport_min, BFD.udp_sport_max,
91 class BFDTestSession(object):
92 """ BFD session as seen from test framework side """
94 def __init__(self, test, interface, detect_mult=3):
96 self.interface = interface
98 'my_discriminator': 0,
99 'desired_min_tx_interval': 100000,
100 'detect_mult': detect_mult,
101 'diag': BFDDiagCode.no_diagnostic,
104 def update(self, **kwargs):
105 self.bfd_values.update(kwargs)
107 def create_packet(self):
108 packet = create_packet(self.interface)
109 self.test.logger.debug("BFD: Creating packet")
110 for name, value in self.bfd_values.iteritems():
111 self.test.logger.debug("BFD: setting packet.%s=%s", name, value)
112 packet[BFD].setfieldval(name, value)
115 def send_packet(self):
116 p = self.create_packet()
117 self.test.logger.debug(ppp("Sending packet:", p))
118 self.test.pg0.add_stream([p])
121 def verify_packet(self, packet):
122 """ Verify correctness of BFD layer. """
124 self.test.assert_equal(bfd.version, 1, "BFD version")
125 self.test.assert_equal(bfd.your_discriminator,
126 self.bfd_values['my_discriminator'],
127 "BFD - your discriminator")
131 class BFDTestCase(VppTestCase):
132 """Bidirectional Forwarding Detection (BFD)"""
136 super(BFDTestCase, cls).setUpClass()
138 cls.create_pg_interfaces([0])
140 cls.pg0.generate_remote_hosts()
141 cls.pg0.configure_ipv4_neighbors()
143 cls.pg0.resolve_arp()
146 super(BFDTestCase, cls).tearDownClass()
150 super(BFDTestCase, self).setUp()
151 self.vapi.want_bfd_events()
152 self.vpp_session = VppBFDUDPSession(
153 self, self.pg0, self.pg0.remote_ip4)
154 self.vpp_session.add_vpp_config()
155 self.vpp_session.admin_up()
156 self.test_session = BFDTestSession(self, self.pg0)
157 self.test_session.update(required_min_rx_interval=100000)
160 self.vapi.collect_events() # clear the event queue
161 if not self.vpp_dead:
162 self.vapi.want_bfd_events(enable_disable=0)
163 self.vpp_session.remove_vpp_config()
164 super(BFDTestCase, self).tearDown()
166 def verify_event(self, event, expected_state):
167 """ Verify correctness of event values. """
169 self.logger.debug("BFD: Event: %s" % repr(e))
170 self.assert_equal(e.bs_index, self.vpp_session.bs_index,
174 self.vpp_session.interface.sw_if_index,
175 "BFD interface index")
177 if self.vpp_session.af == AF_INET6:
179 self.assert_equal(e.is_ipv6, is_ipv6, "is_ipv6")
180 if self.vpp_session.af == AF_INET:
181 self.assert_equal(e.local_addr[:4], self.vpp_session.local_addr_n,
182 "Local IPv4 address")
183 self.assert_equal(e.peer_addr[:4], self.vpp_session.peer_addr_n,
186 self.assert_equal(e.local_addr, self.vpp_session.local_addr_n,
187 "Local IPv6 address")
188 self.assert_equal(e.peer_addr, self.vpp_session.peer_addr_n,
190 self.assert_equal(e.state, expected_state, BFDState)
192 def wait_for_bfd_packet(self, timeout=1):
193 """ wait for BFD packet
195 :param timeout: how long to wait max
197 :returns: tuple (packet, time spent waiting for packet)
199 self.logger.info("BFD: Waiting for BFD packet")
201 p = self.pg0.wait_for_packet(timeout=timeout)
205 raise Exception(ppp("Unexpected or invalid BFD packet:", p))
207 raise Exception(ppp("Unexpected payload in BFD packet:", bfd))
208 verify_ip(self, p, self.pg0.local_ip4, self.pg0.remote_ip4)
210 self.test_session.verify_packet(p)
211 return p, after - before
213 def bfd_session_up(self):
214 self.pg_enable_capture([self.pg0])
215 self.logger.info("BFD: Waiting for slow hello")
216 p, ttp = self.wait_for_bfd_packet()
217 self.logger.info("BFD: Sending Init")
218 self.test_session.update(my_discriminator=randint(0, 40000000),
219 your_discriminator=p[BFD].my_discriminator,
221 self.test_session.send_packet()
222 self.logger.info("BFD: Waiting for event")
223 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
224 self.verify_event(e, expected_state=BFDState.up)
225 self.logger.info("BFD: Session is Up")
226 self.test_session.update(state=BFDState.up)
228 def test_session_up(self):
229 """ bring BFD session up """
230 self.bfd_session_up()
232 def test_slow_timer(self):
233 """ verify slow periodic control frames while session down """
234 self.pg_enable_capture([self.pg0])
236 self.logger.info("BFD: Waiting for %d BFD packets" % expected_packets)
237 self.wait_for_bfd_packet(2)
238 for i in range(expected_packets):
240 self.wait_for_bfd_packet(2)
242 # spec says the range should be <0.75, 1>, allow extra 0.05 margin
243 # to work around timing issues
244 self.assert_in_range(
245 after - before, 0.70, 1.05, "time between slow packets")
248 def test_zero_remote_min_rx(self):
249 """ no packets when zero BFD RemoteMinRxInterval """
250 self.pg_enable_capture([self.pg0])
251 p, timeout = self.wait_for_bfd_packet(2)
252 self.test_session.update(my_discriminator=randint(0, 40000000),
253 your_discriminator=p[BFD].my_discriminator,
255 required_min_rx_interval=0)
256 self.test_session.send_packet()
257 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
258 self.verify_event(e, expected_state=BFDState.up)
261 p = self.pg0.wait_for_packet(timeout=1)
264 raise Exception(ppp("Received unexpected BFD packet:", p))
266 def test_hold_up(self):
267 """ hold BFD session up """
268 self.bfd_session_up()
270 self.wait_for_bfd_packet()
271 self.test_session.send_packet()
273 def test_conn_down(self):
274 """ verify session goes down after inactivity """
275 self.bfd_session_up()
276 self.wait_for_bfd_packet()
277 self.assert_equal(len(self.vapi.collect_events()), 0,
278 "number of bfd events")
279 self.wait_for_bfd_packet()
280 self.assert_equal(len(self.vapi.collect_events()), 0,
281 "number of bfd events")
282 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
283 self.verify_event(e, expected_state=BFDState.down)
285 def test_large_required_min_rx(self):
286 """ large remote RequiredMinRxInterval """
287 self.bfd_session_up()
289 self.test_session.update(required_min_rx_interval=interval)
290 self.test_session.send_packet()
293 while time.time() < now + interval / us_in_sec:
295 p = self.wait_for_bfd_packet()
297 self.logger.error(ppp("Received unexpected packet:", p))
301 self.assert_in_range(count, 0, 1, "number of packets received")
303 def test_immediate_remote_min_rx_reduce(self):
304 """ immediately honor remote min rx reduction """
305 self.vpp_session.remove_vpp_config()
306 self.vpp_session = VppBFDUDPSession(
307 self, self.pg0, self.pg0.remote_ip4, desired_min_tx=10000)
308 self.vpp_session.add_vpp_config()
309 self.test_session.update(desired_min_tx_interval=1000000,
310 required_min_rx_interval=1000000)
311 self.bfd_session_up()
312 self.wait_for_bfd_packet()
314 self.test_session.update(required_min_rx_interval=interval)
315 self.test_session.send_packet()
316 p, ttp = self.wait_for_bfd_packet()
317 # allow extra 10% to work around timing issues, first packet is special
318 self.assert_in_range(ttp, 0, 1.10 * interval / us_in_sec,
319 "time between BFD packets")
320 p, ttp = self.wait_for_bfd_packet()
321 self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec,
322 1.10 * interval / us_in_sec,
323 "time between BFD packets")
324 p, ttp = self.wait_for_bfd_packet()
325 self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec,
326 1.10 * interval / us_in_sec,
327 "time between BFD packets")
329 if __name__ == '__main__':
330 unittest.main(testRunner=VppTestRunner)