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([0])
26 super(BFDAPITestCase, cls).tearDownClass()
29 def test_add_bfd(self):
30 """ create a BFD session """
31 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
32 session.add_vpp_config()
33 self.logger.debug("Session state is %s" % str(session.state))
34 session.remove_vpp_config()
35 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
36 session.add_vpp_config()
37 self.logger.debug("Session state is %s" % str(session.state))
38 session.remove_vpp_config()
40 def test_double_add(self):
41 """ create the same BFD session twice (negative case) """
42 session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
43 session.add_vpp_config()
45 session.add_vpp_config()
47 session.remove_vpp_config()
49 session.remove_vpp_config()
50 raise Exception("Expected failure while adding duplicate "
54 def create_packet(interface, ttl=255, src_port=50000, **kwargs):
55 p = (Ether(src=interface.remote_mac, dst=interface.local_mac) /
56 IP(src=interface.remote_ip4, dst=interface.local_ip4, ttl=ttl) /
57 UDP(sport=src_port, dport=BFD.udp_dport) /
62 def verify_ip(test, packet, local_ip, remote_ip):
63 """ Verify correctness of IP layer. """
65 test.assert_equal(ip.src, local_ip, "IP source address")
66 test.assert_equal(ip.dst, remote_ip, "IP destination address")
67 test.assert_equal(ip.ttl, 255, "IP TTL")
70 def verify_udp(test, packet):
71 """ Verify correctness of UDP layer. """
73 test.assert_equal(udp.dport, BFD.udp_dport, "UDP destination port")
74 test.assert_in_range(udp.sport, BFD.udp_sport_min, BFD.udp_sport_max,
78 class BFDTestSession(object):
79 """ BFD session as seen from test framework side """
81 def __init__(self, test, interface, detect_mult=3):
83 self.interface = interface
85 'my_discriminator': 0,
86 'desired_min_tx_interval': 100000,
87 'detect_mult': detect_mult,
88 'diag': BFDDiagCode.no_diagnostic,
91 def update(self, **kwargs):
92 self.bfd_values.update(kwargs)
94 def create_packet(self):
95 packet = create_packet(self.interface)
96 self.test.logger.debug("BFD: Creating packet")
97 for name, value in self.bfd_values.iteritems():
98 self.test.logger.debug("BFD: setting packet.%s=%s", name, value)
99 packet[BFD].setfieldval(name, value)
102 def send_packet(self):
103 p = self.create_packet()
104 self.test.logger.debug(ppp("Sending packet:", p))
105 self.test.pg0.add_stream([p])
108 def verify_packet(self, packet):
109 """ Verify correctness of BFD layer. """
111 self.test.assert_equal(bfd.version, 1, "BFD version")
112 self.test.assert_equal(bfd.your_discriminator,
113 self.bfd_values['my_discriminator'],
114 "BFD - your discriminator")
117 class BFDTestCase(VppTestCase):
118 """Bidirectional Forwarding Detection (BFD)"""
122 super(BFDTestCase, cls).setUpClass()
124 cls.create_pg_interfaces([0])
126 cls.pg0.generate_remote_hosts()
127 cls.pg0.configure_ipv4_neighbors()
129 cls.pg0.resolve_arp()
132 super(BFDTestCase, cls).tearDownClass()
136 super(BFDTestCase, self).setUp()
137 self.vapi.want_bfd_events()
138 self.vpp_session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
139 self.vpp_session.add_vpp_config()
140 self.vpp_session.admin_up()
141 self.test_session = BFDTestSession(self, self.pg0)
142 self.test_session.update(required_min_rx_interval=100000)
145 self.vapi.collect_events() # clear the event queue
146 if not self.vpp_dead:
147 self.vapi.want_bfd_events(enable_disable=0)
148 self.vpp_session.remove_vpp_config()
149 super(BFDTestCase, self).tearDown()
151 def verify_event(self, event, expected_state):
152 """ Verify correctness of event values. """
154 self.logger.debug("BFD: Event: %s" % repr(e))
155 self.assert_equal(e.bs_index, self.vpp_session.bs_index,
157 self.assert_equal(e.sw_if_index, self.vpp_session.interface.sw_if_index,
158 "BFD interface index")
160 if self.vpp_session.af == AF_INET6:
162 self.assert_equal(e.is_ipv6, is_ipv6, "is_ipv6")
163 if self.vpp_session.af == AF_INET:
164 self.assert_equal(e.local_addr[:4], self.vpp_session.local_addr_n,
165 "Local IPv4 address")
166 self.assert_equal(e.peer_addr[:4], self.vpp_session.peer_addr_n,
169 self.assert_equal(e.local_addr, self.vpp_session.local_addr_n,
170 "Local IPv6 address")
171 self.assert_equal(e.peer_addr, self.vpp_session.peer_addr_n,
173 self.assert_equal(e.state, expected_state, BFDState)
175 def wait_for_bfd_packet(self, timeout=1):
176 """ wait for BFD packet
178 :param timeout: how long to wait max
180 :returns: tuple (packet, time spent waiting for packet)
182 self.logger.info("BFD: Waiting for BFD packet")
184 p = self.pg0.wait_for_packet(timeout=timeout)
188 raise Exception(ppp("Unexpected or invalid BFD packet:", p))
190 raise Exception(ppp("Unexpected payload in BFD packet:", bfd))
191 verify_ip(self, p, self.pg0.local_ip4, self.pg0.remote_ip4)
193 self.test_session.verify_packet(p)
194 return p, after - before
196 def bfd_session_up(self):
197 self.pg_enable_capture([self.pg0])
198 self.logger.info("BFD: Waiting for slow hello")
199 p, ttp = self.wait_for_bfd_packet()
200 self.logger.info("BFD: Sending Init")
201 self.test_session.update(my_discriminator=randint(0, 40000000),
202 your_discriminator=p[BFD].my_discriminator,
204 self.test_session.send_packet()
205 self.logger.info("BFD: Waiting for event")
206 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
207 self.verify_event(e, expected_state=BFDState.up)
208 self.logger.info("BFD: Session is Up")
209 self.test_session.update(state=BFDState.up)
211 def test_session_up(self):
212 """ bring BFD session up """
213 self.bfd_session_up()
215 def test_slow_timer(self):
216 """ verify slow periodic control frames while session down """
217 self.pg_enable_capture([self.pg0])
219 self.logger.info("BFD: Waiting for %d BFD packets" % expected_packets)
220 self.wait_for_bfd_packet(2)
221 for i in range(expected_packets):
223 self.wait_for_bfd_packet(2)
225 # spec says the range should be <0.75, 1>, allow extra 0.05 margin
226 # to work around timing issues
227 self.assert_in_range(
228 after - before, 0.70, 1.05, "time between slow packets")
231 def test_zero_remote_min_rx(self):
232 """ no packets when zero BFD RemoteMinRxInterval """
233 self.pg_enable_capture([self.pg0])
234 p, timeout = self.wait_for_bfd_packet(2)
235 self.test_session.update(my_discriminator=randint(0, 40000000),
236 your_discriminator=p[BFD].my_discriminator,
238 required_min_rx_interval=0)
239 self.test_session.send_packet()
240 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
241 self.verify_event(e, expected_state=BFDState.up)
244 p = self.pg0.wait_for_packet(timeout=1)
247 raise Exception(ppp("Received unexpected BFD packet:", p))
249 def test_hold_up(self):
250 """ hold BFD session up """
251 self.bfd_session_up()
253 self.wait_for_bfd_packet()
254 self.test_session.send_packet()
256 def test_conn_down(self):
257 """ verify session goes down after inactivity """
258 self.bfd_session_up()
259 self.wait_for_bfd_packet()
260 self.assert_equal(len(self.vapi.collect_events()), 0,
261 "number of bfd events")
262 self.wait_for_bfd_packet()
263 self.assert_equal(len(self.vapi.collect_events()), 0,
264 "number of bfd events")
265 e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
266 self.verify_event(e, expected_state=BFDState.down)
268 def test_large_required_min_rx(self):
269 """ large remote RequiredMinRxInterval """
270 self.bfd_session_up()
272 self.test_session.update(required_min_rx_interval=interval)
273 self.test_session.send_packet()
276 while time.time() < now + interval / us_in_sec:
278 p = self.wait_for_bfd_packet()
280 self.logger.error(ppp("Received unexpected packet:", p))
284 self.assert_in_range(count, 0, 1, "number of packets received")
286 def test_immediate_remote_min_rx_reduce(self):
287 """ immediately honor remote min rx reduction """
288 self.vpp_session.remove_vpp_config()
289 self.vpp_session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4,
290 desired_min_tx=10000)
291 self.vpp_session.add_vpp_config()
292 self.test_session.update(desired_min_tx_interval=1000000,
293 required_min_rx_interval=1000000)
294 self.bfd_session_up()
295 self.wait_for_bfd_packet()
297 self.test_session.update(required_min_rx_interval=interval)
298 self.test_session.send_packet()
299 p, ttp = self.wait_for_bfd_packet()
300 # allow extra 10% to work around timing issues, first packet is special
301 self.assert_in_range(ttp, 0, 1.10 * interval / us_in_sec,
302 "time between BFD packets")
303 p, ttp = self.wait_for_bfd_packet()
304 self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec,
305 1.10 * interval / us_in_sec,
306 "time between BFD packets")
307 p, ttp = self.wait_for_bfd_packet()
308 self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec,
309 1.10 * interval / us_in_sec,
310 "time between BFD packets")
312 if __name__ == '__main__':
313 unittest.main(testRunner=VppTestRunner)