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     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()
  40
  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()
  45         try:
  46             session.add_vpp_config()
  47         except:
  48             session.remove_vpp_config()
  49             return
  50         session.remove_vpp_config()
  51         raise Exception("Expected failure while adding duplicate "
  52                         "configuration")
  53
  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)" %
  62                             session1.bs_index)
  63
  64
  65 class BFDTestSession(object):
  66     """ BFD session as seen from test framework side """
  67
  68     def __init__(self, test, interface, af, detect_mult=3):
  69         self.test = test
  70         self.af = af
  71         self.interface = interface
  72         self.udp_sport = 50000
  73         self.bfd_values = {
  74             'my_discriminator': 0,
  75             'desired_min_tx_interval': 100000,
  76             'detect_mult': detect_mult,
  77             'diag': BFDDiagCode.no_diagnostic,
  78         }
  79
  80     def update(self, **kwargs):
  81         self.bfd_values.update(kwargs)
  82
  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,
  89                            hlim=255) /
  90                       UDP(sport=self.udp_sport, dport=BFD.udp_dport) /
  91                       BFD())
  92         else:
  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,
  97                          ttl=255) /
  98                       UDP(sport=self.udp_sport, dport=BFD.udp_dport) /
  99                       BFD())
 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)
 104         return packet
 105
 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])
 110         self.test.pg_start()
 111
 112     def verify_packet(self, packet):
 113         """ Verify correctness of BFD layer. """
 114         bfd = packet[BFD]
 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")
 119
 120
 121 class BFDCommonCode:
 122     """Common code used by both IPv4 and IPv6 Test Cases"""
 123
 124     def tearDown(self):
 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()
 129
 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,
 137                                  state=BFDState.init,
 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)
 145
 146     def verify_ip(self, packet):
 147         """ Verify correctness of IP layer. """
 148         if self.vpp_session.af == AF_INET6:
 149             ip = packet[IPv6]
 150             local_ip = self.pg0.local_ip6
 151             remote_ip = self.pg0.remote_ip6
 152             self.assert_equal(ip.hlim, 255, "IPv6 hop limit")
 153         else:
 154             ip = packet[IP]
 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")
 160
 161     def verify_udp(self, packet):
 162         """ Verify correctness of UDP layer. """
 163         udp = packet[UDP]
 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,
 166                              "UDP source port")
 167
 168     def verify_event(self, event, expected_state):
 169         """ Verify correctness of event values. """
 170         e = event
 171         self.logger.debug("BFD: Event: %s" % repr(e))
 172         self.assert_equal(e.bs_index, self.vpp_session.bs_index,
 173                           "BFD session index")
 174         self.assert_equal(
 175             e.sw_if_index,
 176             self.vpp_session.interface.sw_if_index,
 177             "BFD interface index")
 178         is_ipv6 = 0
 179         if self.vpp_session.af == AF_INET6:
 180             is_ipv6 = 1
 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,
 186                               "Peer IPv4 address")
 187         else:
 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,
 191                               "Peer IPv6 address")
 192         self.assert_equal(e.state, expected_state, BFDState)
 193
 194     def wait_for_bfd_packet(self, timeout=1):
 195         """ wait for BFD packet
 196
 197         :param timeout: how long to wait max
 198
 199         :returns: tuple (packet, time spent waiting for packet)
 200         """
 201         self.logger.info("BFD: Waiting for BFD packet")
 202         before = time.time()
 203         p = self.pg0.wait_for_packet(timeout=timeout)
 204         after = time.time()
 205         self.logger.debug(ppp("Got packet:", p))
 206         bfd = p[BFD]
 207         if bfd is None:
 208             raise Exception(ppp("Unexpected or invalid BFD packet:", p))
 209         if bfd.payload:
 210             raise Exception(ppp("Unexpected payload in BFD packet:", bfd))
 211         self.verify_ip(p)
 212         self.verify_udp(p)
 213         self.test_session.verify_packet(p)
 214         return p, after - before
 215
 216     def test_session_up(self):
 217         """ bring BFD session up """
 218         self.bfd_session_up()
 219
 220     def test_hold_up(self):
 221         """ hold BFD session up """
 222         self.bfd_session_up()
 223         for i in range(5):
 224             self.wait_for_bfd_packet()
 225             self.test_session.send_packet()
 226
 227
 228 class BFD4TestCase(VppTestCase, BFDCommonCode):
 229     """Bidirectional Forwarding Detection (BFD)"""
 230
 231     @classmethod
 232     def setUpClass(cls):
 233         super(BFD4TestCase, cls).setUpClass()
 234         try:
 235             cls.create_pg_interfaces([0])
 236             cls.pg0.config_ip4()
 237             cls.pg0.generate_remote_hosts()
 238             cls.pg0.configure_ipv4_neighbors()
 239             cls.pg0.admin_up()
 240             cls.pg0.resolve_arp()
 241
 242         except Exception:
 243             super(BFD4TestCase, cls).tearDownClass()
 244             raise
 245
 246     def setUp(self):
 247         super(BFD4TestCase, self).setUp()
 248         self.vapi.want_bfd_events()
 249         try:
 250             self.vpp_session = VppBFDUDPSession(self, self.pg0,
 251                                                 self.pg0.remote_ip4)
 252             self.vpp_session.add_vpp_config()
 253             self.vpp_session.admin_up()
 254             self.test_session = BFDTestSession(self, self.pg0, AF_INET)
 255         except:
 256             self.vapi.want_bfd_events(enable_disable=0)
 257             raise
 258
 259     def tearDown(self):
 260         BFDCommonCode.tearDown(self)
 261         super(BFD4TestCase, self).tearDown()
 262
 263     def test_slow_timer(self):
 264         """ verify slow periodic control frames while session down """
 265         self.pg_enable_capture([self.pg0])
 266         expected_packets = 3
 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):
 270             before = time.time()
 271             self.wait_for_bfd_packet(2)
 272             after = time.time()
 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")
 277             before = after
 278
 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,
 285                                  state=BFDState.init,
 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)
 290
 291         try:
 292             p = self.pg0.wait_for_packet(timeout=1)
 293         except:
 294             return
 295         raise Exception(ppp("Received unexpected BFD packet:", p))
 296
 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)
 308
 309     def test_large_required_min_rx(self):
 310         """ large remote RequiredMinRxInterval """
 311         self.bfd_session_up()
 312         interval = 3000000
 313         self.test_session.update(required_min_rx_interval=interval)
 314         self.test_session.send_packet()
 315         now = time.time()
 316         count = 0
 317         while time.time() < now + interval / us_in_sec:
 318             try:
 319                 p = self.wait_for_bfd_packet()
 320                 if count > 1:
 321                     self.logger.error(ppp("Received unexpected packet:", p))
 322                 count += 1
 323             except:
 324                 pass
 325         self.assert_in_range(count, 0, 1, "number of packets received")
 326
 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()
 337         interval = 100000
 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")
 352
 353
 354 class BFD6TestCase(VppTestCase, BFDCommonCode):
 355     """Bidirectional Forwarding Detection (BFD) (IPv6) """
 356
 357     @classmethod
 358     def setUpClass(cls):
 359         super(BFD6TestCase, cls).setUpClass()
 360         try:
 361             cls.create_pg_interfaces([0])
 362             cls.pg0.config_ip6()
 363             cls.pg0.configure_ipv6_neighbors()
 364             cls.pg0.admin_up()
 365             cls.pg0.resolve_ndp()
 366
 367         except Exception:
 368             super(BFD6TestCase, cls).tearDownClass()
 369             raise
 370
 371     def setUp(self):
 372         super(BFD6TestCase, self).setUp()
 373         self.vapi.want_bfd_events()
 374         try:
 375             self.vpp_session = VppBFDUDPSession(self, self.pg0,
 376                                                 self.pg0.remote_ip6,
 377                                                 af=AF_INET6)
 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"))
 382         except:
 383             self.vapi.want_bfd_events(enable_disable=0)
 384             raise
 385
 386     def tearDown(self):
 387         BFDCommonCode.tearDown(self)
 388         super(BFD6TestCase, self).tearDown()
 389
 390 if __name__ == '__main__':
 391     unittest.main(testRunner=VppTestRunner)