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 def create_packet(interface, ttl=255, src_port=50000, **kwargs):
  66     p = (Ether(src=interface.remote_mac, dst=interface.local_mac) /
  67          IP(src=interface.remote_ip4, dst=interface.local_ip4, ttl=ttl) /
  68          UDP(sport=src_port, dport=BFD.udp_dport) /
  69          BFD(*kwargs))
  70     return p
  71
  72
  73 def verify_ip(test, packet, local_ip, remote_ip):
  74     """ Verify correctness of IP layer. """
  75     ip = packet[IP]
  76     test.assert_equal(ip.src, local_ip, "IP source address")
  77     test.assert_equal(ip.dst, remote_ip, "IP destination address")
  78     test.assert_equal(ip.ttl, 255, "IP TTL")
  79
  80
  81 def verify_udp(test, packet):
  82     """ Verify correctness of UDP layer. """
  83     udp = packet[UDP]
  84     test.assert_equal(udp.dport, BFD.udp_dport, "UDP destination port")
  85     test.assert_in_range(udp.sport, BFD.udp_sport_min, BFD.udp_sport_max,
  86                          "UDP source port")
  87
  88
  89 class BFDTestSession(object):
  90     """ BFD session as seen from test framework side """
  91
  92     def __init__(self, test, interface, detect_mult=3):
  93         self.test = test
  94         self.interface = interface
  95         self.bfd_values = {
  96             'my_discriminator': 0,
  97             'desired_min_tx_interval': 100000,
  98             'detect_mult': detect_mult,
  99             'diag': BFDDiagCode.no_diagnostic,
 100         }
 101
 102     def update(self, **kwargs):
 103         self.bfd_values.update(kwargs)
 104
 105     def create_packet(self):
 106         packet = create_packet(self.interface)
 107         self.test.logger.debug("BFD: Creating packet")
 108         for name, value in self.bfd_values.iteritems():
 109             self.test.logger.debug("BFD: setting packet.%s=%s", name, value)
 110             packet[BFD].setfieldval(name, value)
 111         return packet
 112
 113     def send_packet(self):
 114         p = self.create_packet()
 115         self.test.logger.debug(ppp("Sending packet:", p))
 116         self.test.pg0.add_stream([p])
 117         self.test.pg_start()
 118
 119     def verify_packet(self, packet):
 120         """ Verify correctness of BFD layer. """
 121         bfd = packet[BFD]
 122         self.test.assert_equal(bfd.version, 1, "BFD version")
 123         self.test.assert_equal(bfd.your_discriminator,
 124                                self.bfd_values['my_discriminator'],
 125                                "BFD - your discriminator")
 126
 127
 128 @unittest.skip("")
 129 class BFDTestCase(VppTestCase):
 130     """Bidirectional Forwarding Detection (BFD)"""
 131
 132     @classmethod
 133     def setUpClass(cls):
 134         super(BFDTestCase, cls).setUpClass()
 135         try:
 136             cls.create_pg_interfaces([0])
 137             cls.pg0.config_ip4()
 138             cls.pg0.generate_remote_hosts()
 139             cls.pg0.configure_ipv4_neighbors()
 140             cls.pg0.admin_up()
 141             cls.pg0.resolve_arp()
 142
 143         except Exception:
 144             super(BFDTestCase, cls).tearDownClass()
 145             raise
 146
 147     def setUp(self):
 148         super(BFDTestCase, self).setUp()
 149         self.vapi.want_bfd_events()
 150         self.vpp_session = VppBFDUDPSession(
 151             self, self.pg0, self.pg0.remote_ip4)
 152         self.vpp_session.add_vpp_config()
 153         self.vpp_session.admin_up()
 154         self.test_session = BFDTestSession(self, self.pg0)
 155         self.test_session.update(required_min_rx_interval=100000)
 156
 157     def tearDown(self):
 158         self.vapi.collect_events()  # clear the event queue
 159         if not self.vpp_dead:
 160             self.vapi.want_bfd_events(enable_disable=0)
 161             self.vpp_session.remove_vpp_config()
 162         super(BFDTestCase, self).tearDown()
 163
 164     def verify_event(self, event, expected_state):
 165         """ Verify correctness of event values. """
 166         e = event
 167         self.logger.debug("BFD: Event: %s" % repr(e))
 168         self.assert_equal(e.bs_index, self.vpp_session.bs_index,
 169                           "BFD session index")
 170         self.assert_equal(
 171             e.sw_if_index,
 172             self.vpp_session.interface.sw_if_index,
 173             "BFD interface index")
 174         is_ipv6 = 0
 175         if self.vpp_session.af == AF_INET6:
 176             is_ipv6 = 1
 177         self.assert_equal(e.is_ipv6, is_ipv6, "is_ipv6")
 178         if self.vpp_session.af == AF_INET:
 179             self.assert_equal(e.local_addr[:4], self.vpp_session.local_addr_n,
 180                               "Local IPv4 address")
 181             self.assert_equal(e.peer_addr[:4], self.vpp_session.peer_addr_n,
 182                               "Peer IPv4 address")
 183         else:
 184             self.assert_equal(e.local_addr, self.vpp_session.local_addr_n,
 185                               "Local IPv6 address")
 186             self.assert_equal(e.peer_addr, self.vpp_session.peer_addr_n,
 187                               "Peer IPv6 address")
 188         self.assert_equal(e.state, expected_state, BFDState)
 189
 190     def wait_for_bfd_packet(self, timeout=1):
 191         """ wait for BFD packet
 192
 193         :param timeout: how long to wait max
 194
 195         :returns: tuple (packet, time spent waiting for packet)
 196         """
 197         self.logger.info("BFD: Waiting for BFD packet")
 198         before = time.time()
 199         p = self.pg0.wait_for_packet(timeout=timeout)
 200         after = time.time()
 201         bfd = p[BFD]
 202         if bfd is None:
 203             raise Exception(ppp("Unexpected or invalid BFD packet:", p))
 204         if bfd.payload:
 205             raise Exception(ppp("Unexpected payload in BFD packet:", bfd))
 206         verify_ip(self, p, self.pg0.local_ip4, self.pg0.remote_ip4)
 207         verify_udp(self, p)
 208         self.test_session.verify_packet(p)
 209         return p, after - before
 210
 211     def bfd_session_up(self):
 212         self.pg_enable_capture([self.pg0])
 213         self.logger.info("BFD: Waiting for slow hello")
 214         p, ttp = self.wait_for_bfd_packet()
 215         self.logger.info("BFD: Sending Init")
 216         self.test_session.update(my_discriminator=randint(0, 40000000),
 217                                  your_discriminator=p[BFD].my_discriminator,
 218                                  state=BFDState.init)
 219         self.test_session.send_packet()
 220         self.logger.info("BFD: Waiting for event")
 221         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
 222         self.verify_event(e, expected_state=BFDState.up)
 223         self.logger.info("BFD: Session is Up")
 224         self.test_session.update(state=BFDState.up)
 225
 226     def test_session_up(self):
 227         """ bring BFD session up """
 228         self.bfd_session_up()
 229
 230     def test_slow_timer(self):
 231         """ verify slow periodic control frames while session down """
 232         self.pg_enable_capture([self.pg0])
 233         expected_packets = 3
 234         self.logger.info("BFD: Waiting for %d BFD packets" % expected_packets)
 235         self.wait_for_bfd_packet(2)
 236         for i in range(expected_packets):
 237             before = time.time()
 238             self.wait_for_bfd_packet(2)
 239             after = time.time()
 240             # spec says the range should be <0.75, 1>, allow extra 0.05 margin
 241             # to work around timing issues
 242             self.assert_in_range(
 243                 after - before, 0.70, 1.05, "time between slow packets")
 244             before = after
 245
 246     def test_zero_remote_min_rx(self):
 247         """ no packets when zero BFD RemoteMinRxInterval """
 248         self.pg_enable_capture([self.pg0])
 249         p, timeout = self.wait_for_bfd_packet(2)
 250         self.test_session.update(my_discriminator=randint(0, 40000000),
 251                                  your_discriminator=p[BFD].my_discriminator,
 252                                  state=BFDState.init,
 253                                  required_min_rx_interval=0)
 254         self.test_session.send_packet()
 255         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
 256         self.verify_event(e, expected_state=BFDState.up)
 257
 258         try:
 259             p = self.pg0.wait_for_packet(timeout=1)
 260         except:
 261             return
 262         raise Exception(ppp("Received unexpected BFD packet:", p))
 263
 264     def test_hold_up(self):
 265         """ hold BFD session up """
 266         self.bfd_session_up()
 267         for i in range(5):
 268             self.wait_for_bfd_packet()
 269             self.test_session.send_packet()
 270
 271     def test_conn_down(self):
 272         """ verify session goes down after inactivity """
 273         self.bfd_session_up()
 274         self.wait_for_bfd_packet()
 275         self.assert_equal(len(self.vapi.collect_events()), 0,
 276                           "number of bfd events")
 277         self.wait_for_bfd_packet()
 278         self.assert_equal(len(self.vapi.collect_events()), 0,
 279                           "number of bfd events")
 280         e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
 281         self.verify_event(e, expected_state=BFDState.down)
 282
 283     def test_large_required_min_rx(self):
 284         """ large remote RequiredMinRxInterval """
 285         self.bfd_session_up()
 286         interval = 3000000
 287         self.test_session.update(required_min_rx_interval=interval)
 288         self.test_session.send_packet()
 289         now = time.time()
 290         count = 0
 291         while time.time() < now + interval / us_in_sec:
 292             try:
 293                 p = self.wait_for_bfd_packet()
 294                 if count > 1:
 295                     self.logger.error(ppp("Received unexpected packet:", p))
 296                 count += 1
 297             except:
 298                 pass
 299         self.assert_in_range(count, 0, 1, "number of packets received")
 300
 301     def test_immediate_remote_min_rx_reduce(self):
 302         """ immediately honor remote min rx reduction """
 303         self.vpp_session.remove_vpp_config()
 304         self.vpp_session = VppBFDUDPSession(
 305             self, self.pg0, self.pg0.remote_ip4, desired_min_tx=10000)
 306         self.vpp_session.add_vpp_config()
 307         self.test_session.update(desired_min_tx_interval=1000000,
 308                                  required_min_rx_interval=1000000)
 309         self.bfd_session_up()
 310         self.wait_for_bfd_packet()
 311         interval = 100000
 312         self.test_session.update(required_min_rx_interval=interval)
 313         self.test_session.send_packet()
 314         p, ttp = self.wait_for_bfd_packet()
 315         # allow extra 10% to work around timing issues, first packet is special
 316         self.assert_in_range(ttp, 0, 1.10 * interval / us_in_sec,
 317                              "time between BFD packets")
 318         p, ttp = self.wait_for_bfd_packet()
 319         self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec,
 320                              1.10 * interval / us_in_sec,
 321                              "time between BFD packets")
 322         p, ttp = self.wait_for_bfd_packet()
 323         self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec,
 324                              1.10 * interval / us_in_sec,
 325                              "time between BFD packets")
 326
 327 if __name__ == '__main__':
 328     unittest.main(testRunner=VppTestRunner)