from util import ppp
-class BFDCLITestCase(VppTestCase):
- """Bidirectional Forwarding Detection (BFD) - CLI"""
+class BFDAPITestCase(VppTestCase):
+ """Bidirectional Forwarding Detection (BFD) - API"""
@classmethod
def setUpClass(cls):
- super(BFDCLITestCase, cls).setUpClass()
+ super(BFDAPITestCase, cls).setUpClass()
try:
cls.create_pg_interfaces([0])
cls.pg0.resolve_arp()
except Exception:
- super(BFDCLITestCase, cls).tearDownClass()
+ super(BFDAPITestCase, cls).tearDownClass()
raise
def test_add_bfd(self):
self.interface = interface
self.bfd_values = {
'my_discriminator': 0,
- 'desired_min_tx_interval': 500000,
+ 'desired_min_tx_interval': 100000,
'detect_mult': detect_mult,
'diag': BFDDiagCode.no_diagnostic,
}
def setUpClass(cls):
super(BFDTestCase, cls).setUpClass()
try:
- cls.create_pg_interfaces([0, 1])
+ cls.create_pg_interfaces([0])
cls.pg0.config_ip4()
cls.pg0.generate_remote_hosts()
cls.pg0.configure_ipv4_neighbors()
raise
def setUp(self):
+ super(BFDTestCase, self).setUp()
self.vapi.want_bfd_events()
self.vpp_session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
self.vpp_session.add_vpp_config()
def tearDown(self):
self.vapi.want_bfd_events(enable_disable=0)
+ self.vapi.collect_events() # clear the event queue
if not self.vpp_dead:
self.vpp_session.remove_vpp_config()
super(BFDTestCase, self).tearDown()
def verify_event(self, event, expected_state):
""" Verify correctness of event values. """
e = event
- self.logger.debug("Event: %s" % repr(e))
+ self.logger.debug("BFD: Event: %s" % repr(e))
self.assert_equal(e.bs_index, self.vpp_session.bs_index,
"BFD session index")
self.assert_equal(e.sw_if_index, self.vpp_session.interface.sw_if_index,
self.assert_equal(e.state, expected_state, BFDState)
def wait_for_bfd_packet(self, timeout=1):
+ self.logger.info("BFD: Waiting for BFD packet")
p = self.pg0.wait_for_packet(timeout=timeout)
bfd = p[BFD]
if bfd is None:
return p
def test_slow_timer(self):
- """ Slow timer """
-
+ """ verify slow periodic control frames while session down """
self.pg_enable_capture([self.pg0])
- expected_packets = 10
- self.logger.info("Waiting for %d BFD packets" % expected_packets)
- self.wait_for_bfd_packet()
+ expected_packets = 3
+ self.logger.info("BFD: Waiting for %d BFD packets" % expected_packets)
+ self.wait_for_bfd_packet(2)
for i in range(expected_packets):
before = time.time()
- self.wait_for_bfd_packet()
+ self.wait_for_bfd_packet(2)
after = time.time()
+ # spec says the range should be <0.75, 1>, allow extra 0.05 margin
+ # to work around timing issues
self.assert_in_range(
- after - before, 0.75, 1, "time between slow packets")
+ after - before, 0.70, 1.05, "time between slow packets")
before = after
def test_zero_remote_min_rx(self):
- """ Zero RemoteMinRxInterval """
+ """ no packets when zero BFD RemoteMinRxInterval """
self.pg_enable_capture([self.pg0])
- p = self.wait_for_bfd_packet()
+ p = self.wait_for_bfd_packet(2)
self.test_session.update(my_discriminator=randint(0, 40000000),
your_discriminator=p[BFD].my_discriminator,
state=BFDState.init,
return
raise Exception(ppp("Received unexpected BFD packet:", p))
- def bfd_conn_up(self):
+ def bfd_session_up(self):
self.pg_enable_capture([self.pg0])
- self.logger.info("Waiting for slow hello")
- p = self.wait_for_bfd_packet()
- self.logger.info("Sending Init")
+ self.logger.info("BFD: Waiting for slow hello")
+ p = self.wait_for_bfd_packet(2)
+ self.logger.info("BFD: Sending Init")
self.test_session.update(my_discriminator=randint(0, 40000000),
your_discriminator=p[BFD].my_discriminator,
state=BFDState.init,
- required_min_rx_interval=500000)
+ required_min_rx_interval=100000)
self.test_session.send_packet()
- self.logger.info("Waiting for event")
+ self.logger.info("BFD: Waiting for event")
e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
self.verify_event(e, expected_state=BFDState.up)
- self.logger.info("Session is Up")
+ self.logger.info("BFD: Session is Up")
self.test_session.update(state=BFDState.up)
- def test_conn_up(self):
- """ Basic connection up """
- self.bfd_conn_up()
+ def test_session_up(self):
+ """ bring BFD session up """
+ self.bfd_session_up()
def test_hold_up(self):
- """ Hold BFD up """
- self.bfd_conn_up()
+ """ hold BFD session up """
+ self.bfd_session_up()
for i in range(5):
self.wait_for_bfd_packet()
self.test_session.send_packet()
def test_conn_down(self):
- """ Session down after inactivity """
- self.bfd_conn_up()
+ """ verify session goes down after inactivity """
+ self.bfd_session_up()
self.wait_for_bfd_packet()
- self.assert_equal(
- 0, len(self.vapi.collect_events()),
- "number of bfd events")
+ self.assert_equal(len(self.vapi.collect_events()), 0,
+ "number of bfd events")
self.wait_for_bfd_packet()
- self.assert_equal(
- 0, len(self.vapi.collect_events()),
- "number of bfd events")
+ self.assert_equal(len(self.vapi.collect_events()), 0,
+ "number of bfd events")
e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
self.verify_event(e, expected_state=BFDState.down)
- @unittest.skip("this test is not working yet")
def test_large_required_min_rx(self):
- self.bfd_conn_up()
- interval = 5000000
+ """ large remote RequiredMinRxInterval """
+ self.bfd_session_up()
+ interval = 3000000
self.test_session.update(required_min_rx_interval=interval)
self.test_session.send_packet()
now = time.time()
- count = 1
+ count = 0
while time.time() < now + interval / 1000000:
try:
p = self.wait_for_bfd_packet()
count += 1
except:
pass
- self.assert_equal(count, 1, "number of packets received")
+ self.assert_in_range(count, 0, 1, "number of packets received")
if __name__ == '__main__':
Code Review / vpp.git / blobdiff