test/test_bfd.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 import time
   5 from random import randint
   6 from bfd import *
   7 from framework import *
   8 from util import ppp
   9
  10 us_in_sec = 1000000
  11
  12
  13 class BFDAPITestCase(VppTestCase):
  14     """Bidirectional Forwarding Detection (BFD) - API"""
  15
  16     @classmethod
  17     def setUpClass(cls):
  18         super(BFDAPITestCase, cls).setUpClass()
  19
  20         try:
  21             cls.create_pg_interfaces(range(2))
  22             for i in cls.pg_interfaces:
  23                 i.config_ip4()
  24                 i.resolve_arp()
  25
  26         except Exception:
  27             super(BFDAPITestCase, cls).tearDownClass()
  28             raise
  29
  30
  31
  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()
  42
  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()
  47         try:
  48             session.add_vpp_config()
  49         except:
  50             session.remove_vpp_config()
  51             return
  52         session.remove_vpp_config()
  53         raise Exception("Expected failure while adding duplicate "
  54                         "configuration")
  55
  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)" %
  64                             session1.bs_index)
  65
  66
  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) /
  71          BFD(*kwargs))
  72     return p
  73
  74
  75 def verify_ip(test, packet, local_ip, remote_ip):
  76     """ Verify correctness of IP layer. """
  77     ip = packet[IP]
  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")
  81
  82
  83 def verify_udp(test, packet):
  84     """ Verify correctness of UDP layer. """
  85     udp = packet[UDP]
  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,
  88                          "UDP source port")
  89
  90
  91 class BFDTestSession(object):
  92     """ BFD session as seen from test framework side """
  93
  94     def __init__(self, test, interface, detect_mult=3):
  95         self.test = test
  96         self.interface = interface
  97         self.bfd_values = {
  98             'my_discriminator': 0,
  99             'desired_min_tx_interval': 100000,
 100             'detect_mult': detect_mult,
 101             'diag': BFDDiagCode.no_diagnostic,
 102         }
 103
 104     def update(self, **kwargs):
 105         self.bfd_values.update(kwargs)
 106
 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)
 113         return packet
 114
 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])
 119         self.test.pg_start()
 120
 121     def verify_packet(self, packet):
 122         """ Verify correctness of BFD layer. """
 123         bfd = packet[BFD]
 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")
 128
 129
 130 @unittest.skip("")
 131 class BFDTestCase(VppTestCase):
 132     """Bidirectional Forwarding Detection (BFD)"""
 133
 134     @classmethod
 135     def setUpClass(cls):
 136         super(BFDTestCase, cls).setUpClass()
 137         try:
 138             cls.create_pg_interfaces([0])
 139             cls.pg0.config_ip4()
 140             cls.pg0.generate_remote_hosts()
 141             cls.pg0.configure_ipv4_neighbors()
 142             cls.pg0.admin_up()
 143             cls.pg0.resolve_arp()
 144
 145         except Exception:
 146             super(BFDTestCase, cls).tearDownClass()
 147             raise
 148
 149     def setUp(self):
 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)
 158
 159     def tearDown(self):
 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()
 165
 166     def verify_event(self, event, expected_state):
 167         """ Verify correctness of event values. """
 168         e = event
 169         self.logger.debug("BFD: Event: %s" % repr(e))
 170         self.assert_equal(e.bs_index, self.vpp_session.bs_index,
 171                           "BFD session index")
 172         self.assert_equal(
 173             e.sw_if_index,
 174             self.vpp_session.interface.sw_if_index,
 175             "BFD interface index")
 176         is_ipv6 = 0
 177         if self.vpp_session.af == AF_INET6:
 178             is_ipv6 = 1
 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,
 184                               "Peer IPv4 address")
 185         else:
 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,
 189                               "Peer IPv6 address")
 190         self.assert_equal(e.state, expected_state, BFDState)
 191
 192     def wait_for_bfd_packet(self, timeout=1):
 193         """ wait for BFD packet
 194
 195         :param timeout: how long to wait max
 196
 197         :returns: tuple (packet, time spent waiting for packet)
 198         """
 199         self.logger.info("BFD: Waiting for BFD packet")
 200         before = time.time()
 201         p = self.pg0.wait_for_packet(timeout=timeout)
 202         after = time.time()
 203         bfd = p[BFD]
 204         if bfd is None:
 205             raise Exception(ppp("Unexpected or invalid BFD packet:", p))
 206         if bfd.payload:
 207             raise Exception(ppp("Unexpected payload in BFD packet:", bfd))
 208         verify_ip(self, p, self.pg0.local_ip4, self.pg0.remote_ip4)
 209         verify_udp(self, p)
 210         self.test_session.verify_packet(p)
 211         return p, after - before
 212
 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,
 220                                  state=BFDState.init)
 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)
 227
 228     def test_session_up(self):
 229         """ bring BFD session up """
 230         self.bfd_session_up()
 231
 232     def test_slow_timer(self):
 233         """ verify slow periodic control frames while session down """
 234         self.pg_enable_capture([self.pg0])
 235         expected_packets = 3
 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):
 239             before = time.time()
 240             self.wait_for_bfd_packet(2)
 241             after = time.time()
 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")
 246             before = after
 247
 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,
 254                                  state=BFDState.init,
 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)
 259
 260         try:
 261             p = self.pg0.wait_for_packet(timeout=1)
 262         except:
 263             return
 264         raise Exception(ppp("Received unexpected BFD packet:", p))
 265
 266     def test_hold_up(self):
 267         """ hold BFD session up """
 268         self.bfd_session_up()
 269         for i in range(5):
 270             self.wait_for_bfd_packet()
 271             self.test_session.send_packet()
 272
 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)
 284
 285     def test_large_required_min_rx(self):
 286         """ large remote RequiredMinRxInterval """
 287         self.bfd_session_up()
 288         interval = 3000000
 289         self.test_session.update(required_min_rx_interval=interval)
 290         self.test_session.send_packet()
 291         now = time.time()
 292         count = 0
 293         while time.time() < now + interval / us_in_sec:
 294             try:
 295                 p = self.wait_for_bfd_packet()
 296                 if count > 1:
 297                     self.logger.error(ppp("Received unexpected packet:", p))
 298                 count += 1
 299             except:
 300                 pass
 301         self.assert_in_range(count, 0, 1, "number of packets received")
 302
 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()
 313         interval = 100000
 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")
 328
 329 if __name__ == '__main__':
 330     unittest.main(testRunner=VppTestRunner)