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([0])
  22             cls.pg0.config_ip4()
  23             cls.pg0.resolve_arp()
  24
  25         except Exception:
  26             super(BFDAPITestCase, cls).tearDownClass()
  27             raise
  28
  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()
  39
  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()
  44         try:
  45             session.add_vpp_config()
  46         except:
  47             session.remove_vpp_config()
  48             return
  49         session.remove_vpp_config()
  50         raise Exception("Expected failure while adding duplicate "
  51                         "configuration")
  52
  53
  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) /
  58          BFD(*kwargs))
  59     return p
  60
  61
  62 def verify_ip(test, packet, local_ip, remote_ip):
  63     """ Verify correctness of IP layer. """
  64     ip = packet[IP]
  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")
  68
  69
  70 def verify_udp(test, packet):
  71     """ Verify correctness of UDP layer. """
  72     udp = packet[UDP]
  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,
  75                          "UDP source port")
  76
  77
  78 class BFDTestSession(object):
  79     """ BFD session as seen from test framework side """
  80
  81     def __init__(self, test, interface, detect_mult=3):
  82         self.test = test
  83         self.interface = interface
  84         self.bfd_values = {
  85             'my_discriminator': 0,
  86             'desired_min_tx_interval': 100000,
  87             'detect_mult': detect_mult,
  88             'diag': BFDDiagCode.no_diagnostic,
  89         }
  90
  91     def update(self, **kwargs):
  92         self.bfd_values.update(kwargs)
  93
  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)
 100         return packet
 101
 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])
 106         self.test.pg_start()
 107
 108     def verify_packet(self, packet):
 109         """ Verify correctness of BFD layer. """
 110         bfd = packet[BFD]
 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")
 115
 116
 117 class BFDTestCase(VppTestCase):
 118     """Bidirectional Forwarding Detection (BFD)"""
 119
 120     @classmethod
 121     def setUpClass(cls):
 122         super(BFDTestCase, cls).setUpClass()
 123         try:
 124             cls.create_pg_interfaces([0])
 125             cls.pg0.config_ip4()
 126             cls.pg0.generate_remote_hosts()
 127             cls.pg0.configure_ipv4_neighbors()
 128             cls.pg0.admin_up()
 129             cls.pg0.resolve_arp()
 130
 131         except Exception:
 132             super(BFDTestCase, cls).tearDownClass()
 133             raise
 134
 135     def setUp(self):
 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)
 143
 144     def tearDown(self):
 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()
 150
 151     def verify_event(self, event, expected_state):
 152         """ Verify correctness of event values. """
 153         e = event
 154         self.logger.debug("BFD: Event: %s" % repr(e))
 155         self.assert_equal(e.bs_index, self.vpp_session.bs_index,
 156                           "BFD session index")
 157         self.assert_equal(e.sw_if_index, self.vpp_session.interface.sw_if_index,
 158                           "BFD interface index")
 159         is_ipv6 = 0
 160         if self.vpp_session.af == AF_INET6:
 161             is_ipv6 = 1
 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,
 167                               "Peer IPv4 address")
 168         else:
 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,
 172                               "Peer IPv6 address")
 173         self.assert_equal(e.state, expected_state, BFDState)
 174
 175     def wait_for_bfd_packet(self, timeout=1):
 176         """ wait for BFD packet
 177
 178         :param timeout: how long to wait max
 179
 180         :returns: tuple (packet, time spent waiting for packet)
 181         """
 182         self.logger.info("BFD: Waiting for BFD packet")
 183         before = time.time()
 184         p = self.pg0.wait_for_packet(timeout=timeout)
 185         after = time.time()
 186         bfd = p[BFD]
 187         if bfd is None:
 188             raise Exception(ppp("Unexpected or invalid BFD packet:", p))
 189         if bfd.payload:
 190             raise Exception(ppp("Unexpected payload in BFD packet:", bfd))
 191         verify_ip(self, p, self.pg0.local_ip4, self.pg0.remote_ip4)
 192         verify_udp(self, p)
 193         self.test_session.verify_packet(p)
 194         return p, after - before
 195
 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,
 203                                  state=BFDState.init)
 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)
 210
 211     def test_session_up(self):
 212         """ bring BFD session up """
 213         self.bfd_session_up()
 214
 215     def test_slow_timer(self):
 216         """ verify slow periodic control frames while session down """
 217         self.pg_enable_capture([self.pg0])
 218         expected_packets = 3
 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):
 222             before = time.time()
 223             self.wait_for_bfd_packet(2)
 224             after = time.time()
 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")
 229             before = after
 230
 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,
 237                                  state=BFDState.init,
 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)
 242
 243         try:
 244             p = self.pg0.wait_for_packet(timeout=1)
 245         except:
 246             return
 247         raise Exception(ppp("Received unexpected BFD packet:", p))
 248
 249     def test_hold_up(self):
 250         """ hold BFD session up """
 251         self.bfd_session_up()
 252         for i in range(5):
 253             self.wait_for_bfd_packet()
 254             self.test_session.send_packet()
 255
 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)
 267
 268     def test_large_required_min_rx(self):
 269         """ large remote RequiredMinRxInterval """
 270         self.bfd_session_up()
 271         interval = 3000000
 272         self.test_session.update(required_min_rx_interval=interval)
 273         self.test_session.send_packet()
 274         now = time.time()
 275         count = 0
 276         while time.time() < now + interval / us_in_sec:
 277             try:
 278                 p = self.wait_for_bfd_packet()
 279                 if count > 1:
 280                     self.logger.error(ppp("Received unexpected packet:", p))
 281                 count += 1
 282             except:
 283                 pass
 284         self.assert_in_range(count, 0, 1, "number of packets received")
 285
 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()
 296         interval = 100000
 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")
 311
 312 if __name__ == '__main__':
 313     unittest.main(testRunner=VppTestRunner)