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
  46         with self.vapi.expect_negative_api_retval():
  47             session.add_vpp_config()
  48
  49         session.remove_vpp_config()
  50
  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)" %
  59                             session1.bs_index)
  60
  61
  62 class BFDTestSession(object):
  63     """ BFD session as seen from test framework side """
  64
  65     def __init__(self, test, interface, af, detect_mult=3):
  66         self.test = test
  67         self.af = af
  68         self.interface = interface
  69         self.udp_sport = 50000
  70         self.bfd_values = {
  71             'my_discriminator': 0,
  72             'desired_min_tx_interval': 100000,
  73             'detect_mult': detect_mult,
  74             'diag': BFDDiagCode.no_diagnostic,
  75         }
  76
  77     def update(self, **kwargs):
  78         self.bfd_values.update(kwargs)
  79
  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,
  86                            hlim=255) /
  87                       UDP(sport=self.udp_sport, dport=BFD.udp_dport) /
  88                       BFD())
  89         else:
  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,
  94                          ttl=255) /
  95                       UDP(sport=self.udp_sport, dport=BFD.udp_dport) /
  96                       BFD())
  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)
 101         return packet
 102
 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])
 107         self.test.pg_start()
 108
 109     def verify_packet(self, packet):
 110         """ Verify correctness of BFD layer. """
 111         bfd = packet[BFD]
 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")
 116
 117
 118 class BFDCommonCode:
 119     """Common code used by both IPv4 and IPv6 Test Cases"""
 120
 121     def tearDown(self):
 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()
 126
 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,
 134                                  state=BFDState.init,
 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)
 142
 143     def verify_ip(self, packet):
 144         """ Verify correctness of IP layer. """
 145         if self.vpp_session.af == AF_INET6:
 146             ip = packet[IPv6]
 147             local_ip = self.pg0.local_ip6
 148             remote_ip = self.pg0.remote_ip6
 149             self.assert_equal(ip.hlim, 255, "IPv6 hop limit")
 150         else:
 151             ip = packet[IP]
 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")
 157
 158     def verify_udp(self, packet):
 159         """ Verify correctness of UDP layer. """
 160         udp = packet[UDP]
 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,
 163                              "UDP source port")
 164
 165     def verify_event(self, event, expected_state):
 166         """ Verify correctness of event values. """
 167         e = event
 168         self.logger.debug("BFD: Event: %s" % repr(e))
 169         self.assert_equal(e.bs_index, self.vpp_session.bs_index,
 170                           "BFD session index")
 171         self.assert_equal(
 172             e.sw_if_index,
 173             self.vpp_session.interface.sw_if_index,
 174             "BFD interface index")
 175         is_ipv6 = 0
 176         if self.vpp_session.af == AF_INET6:
 177             is_ipv6 = 1
 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,
 183                               "Peer IPv4 address")
 184         else:
 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,
 188                               "Peer IPv6 address")
 189         self.assert_equal(e.state, expected_state, BFDState)
 190
 191     def wait_for_bfd_packet(self, timeout=1):
 192         """ wait for BFD packet
 193
 194         :param timeout: how long to wait max
 195
 196         :returns: tuple (packet, time spent waiting for packet)
 197         """
 198         self.logger.info("BFD: Waiting for BFD packet")
 199         before = time.time()
 200         p = self.pg0.wait_for_packet(timeout=timeout)
 201         after = time.time()
 202         self.logger.debug(ppp("Got packet:", p))
 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         self.verify_ip(p)
 209         self.verify_udp(p)
 210         self.test_session.verify_packet(p)
 211         return p, after - before
 212
 213     def test_session_up(self):
 214         """ bring BFD session up """
 215         self.bfd_session_up()
 216
 217     def test_hold_up(self):
 218         """ hold BFD session up """
 219         self.bfd_session_up()
 220         for i in range(5):
 221             self.wait_for_bfd_packet()
 222             self.test_session.send_packet()
 223
 224
 225 class BFD4TestCase(VppTestCase, BFDCommonCode):
 226     """Bidirectional Forwarding Detection (BFD)"""
 227
 228     @classmethod
 229     def setUpClass(cls):
 230         super(BFD4TestCase, cls).setUpClass()
 231         try:
 232             cls.create_pg_interfaces([0])
 233             cls.pg0.config_ip4()
 234             cls.pg0.generate_remote_hosts()
 235             cls.pg0.configure_ipv4_neighbors()
 236             cls.pg0.admin_up()
 237             cls.pg0.resolve_arp()
 238
 239         except Exception:
 240             super(BFD4TestCase, cls).tearDownClass()
 241             raise
 242
 243     def setUp(self):
 244         super(BFD4TestCase, self).setUp()
 245         self.vapi.want_bfd_events()
 246         try:
 247             self.vpp_session = VppBFDUDPSession(self, self.pg0,
 248                                                 self.pg0.remote_ip4)
 249             self.vpp_session.add_vpp_config()
 250             self.vpp_session.admin_up()
 251             self.test_session = BFDTestSession(self, self.pg0, AF_INET)
 252         except:
 253             self.vapi.want_bfd_events(enable_disable=0)
 254             raise
 255
 256     def tearDown(self):
 257         BFDCommonCode.tearDown(self)
 258         super(BFD4TestCase, self).tearDown()
 259
 260     def test_slow_timer(self):
 261         """ verify slow periodic control frames while session down """
 262         self.pg_enable_capture([self.pg0])
 263         expected_packets = 3
 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):
 267             before = time.time()
 268             self.wait_for_bfd_packet(2)
 269             after = time.time()
 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")
 274             before = after
 275
 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,
 282                                  state=BFDState.init,
 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)
 287
 288         try:
 289             p = self.pg0.wait_for_packet(timeout=1)
 290         except:
 291             return
 292         raise Exception(ppp("Received unexpected BFD packet:", p))
 293
 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)
 305
 306     def test_large_required_min_rx(self):
 307         """ large remote RequiredMinRxInterval """
 308         self.bfd_session_up()
 309         interval = 3000000
 310         self.test_session.update(required_min_rx_interval=interval)
 311         self.test_session.send_packet()
 312         now = time.time()
 313         count = 0
 314         while time.time() < now + interval / us_in_sec:
 315             try:
 316                 p = self.wait_for_bfd_packet()
 317                 if count > 1:
 318                     self.logger.error(ppp("Received unexpected packet:", p))
 319                 count += 1
 320             except:
 321                 pass
 322         self.assert_in_range(count, 0, 1, "number of packets received")
 323
 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()
 334         interval = 100000
 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")
 349
 350
 351 class BFD6TestCase(VppTestCase, BFDCommonCode):
 352     """Bidirectional Forwarding Detection (BFD) (IPv6) """
 353
 354     @classmethod
 355     def setUpClass(cls):
 356         super(BFD6TestCase, cls).setUpClass()
 357         try:
 358             cls.create_pg_interfaces([0])
 359             cls.pg0.config_ip6()
 360             cls.pg0.configure_ipv6_neighbors()
 361             cls.pg0.admin_up()
 362             cls.pg0.resolve_ndp()
 363
 364         except Exception:
 365             super(BFD6TestCase, cls).tearDownClass()
 366             raise
 367
 368     def setUp(self):
 369         super(BFD6TestCase, self).setUp()
 370         self.vapi.want_bfd_events()
 371         try:
 372             self.vpp_session = VppBFDUDPSession(self, self.pg0,
 373                                                 self.pg0.remote_ip6,
 374                                                 af=AF_INET6)
 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"))
 379         except:
 380             self.vapi.want_bfd_events(enable_disable=0)
 381             raise
 382
 383     def tearDown(self):
 384         BFDCommonCode.tearDown(self)
 385         super(BFD6TestCase, self).tearDown()
 386
 387 if __name__ == '__main__':
 388     unittest.main(testRunner=VppTestRunner)