dab231a11e
Perform accounting of expected packets based on created packet infos. Use this accounting info to automatically expect (and verify) the correct number of packets to be captured. Automatically retry the read of the capture file if scapy raises an exception while doing so to handle rare cases when capture file is read while only partially written during busy wait. Don't fail assert_nothing_captured if only junk packets arrived. Change-Id: I16ec2e9410ef510d313ec16b7e13c57d0b2a63f5 Signed-off-by: Klement Sekera <ksekera@cisco.com>
276 lines
10 KiB
Python
276 lines
10 KiB
Python
#!/usr/bin/env python
|
|
|
|
import unittest
|
|
import time
|
|
from random import randint
|
|
from bfd import *
|
|
from framework import *
|
|
from util import ppp
|
|
|
|
|
|
class BFDAPITestCase(VppTestCase):
|
|
"""Bidirectional Forwarding Detection (BFD) - API"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(BFDAPITestCase, cls).setUpClass()
|
|
|
|
try:
|
|
cls.create_pg_interfaces([0])
|
|
cls.pg0.config_ip4()
|
|
cls.pg0.resolve_arp()
|
|
|
|
except Exception:
|
|
super(BFDAPITestCase, cls).tearDownClass()
|
|
raise
|
|
|
|
def test_add_bfd(self):
|
|
""" create a BFD session """
|
|
session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
|
|
session.add_vpp_config()
|
|
self.logger.debug("Session state is %s" % str(session.state))
|
|
session.remove_vpp_config()
|
|
session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
|
|
session.add_vpp_config()
|
|
self.logger.debug("Session state is %s" % str(session.state))
|
|
session.remove_vpp_config()
|
|
|
|
def test_double_add(self):
|
|
""" create the same BFD session twice (negative case) """
|
|
session = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4)
|
|
session.add_vpp_config()
|
|
try:
|
|
session.add_vpp_config()
|
|
except:
|
|
session.remove_vpp_config()
|
|
return
|
|
session.remove_vpp_config()
|
|
raise Exception("Expected failure while adding duplicate "
|
|
"configuration")
|
|
|
|
|
|
def create_packet(interface, ttl=255, src_port=50000, **kwargs):
|
|
p = (Ether(src=interface.remote_mac, dst=interface.local_mac) /
|
|
IP(src=interface.remote_ip4, dst=interface.local_ip4, ttl=ttl) /
|
|
UDP(sport=src_port, dport=BFD.udp_dport) /
|
|
BFD(*kwargs))
|
|
return p
|
|
|
|
|
|
def verify_ip(test, packet, local_ip, remote_ip):
|
|
""" Verify correctness of IP layer. """
|
|
ip = packet[IP]
|
|
test.assert_equal(ip.src, local_ip, "IP source address")
|
|
test.assert_equal(ip.dst, remote_ip, "IP destination address")
|
|
test.assert_equal(ip.ttl, 255, "IP TTL")
|
|
|
|
|
|
def verify_udp(test, packet):
|
|
""" Verify correctness of UDP layer. """
|
|
udp = packet[UDP]
|
|
test.assert_equal(udp.dport, BFD.udp_dport, "UDP destination port")
|
|
test.assert_in_range(udp.sport, BFD.udp_sport_min, BFD.udp_sport_max,
|
|
"UDP source port")
|
|
|
|
|
|
class BFDTestSession(object):
|
|
|
|
def __init__(self, test, interface, detect_mult=3):
|
|
self.test = test
|
|
self.interface = interface
|
|
self.bfd_values = {
|
|
'my_discriminator': 0,
|
|
'desired_min_tx_interval': 100000,
|
|
'detect_mult': detect_mult,
|
|
'diag': BFDDiagCode.no_diagnostic,
|
|
}
|
|
|
|
def update(self, **kwargs):
|
|
self.bfd_values.update(kwargs)
|
|
|
|
def create_packet(self):
|
|
packet = create_packet(self.interface)
|
|
for name, value in self.bfd_values.iteritems():
|
|
packet[BFD].setfieldval(name, value)
|
|
return packet
|
|
|
|
def send_packet(self):
|
|
p = self.create_packet()
|
|
self.test.logger.debug(ppp("Sending packet:", p))
|
|
self.test.pg0.add_stream([p])
|
|
self.test.pg_start()
|
|
|
|
def verify_packet(self, packet):
|
|
""" Verify correctness of BFD layer. """
|
|
bfd = packet[BFD]
|
|
self.test.assert_equal(bfd.version, 1, "BFD version")
|
|
self.test.assert_equal(bfd.your_discriminator,
|
|
self.bfd_values['my_discriminator'],
|
|
"BFD - your discriminator")
|
|
|
|
|
|
class BFDTestCase(VppTestCase):
|
|
"""Bidirectional Forwarding Detection (BFD)"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(BFDTestCase, cls).setUpClass()
|
|
try:
|
|
cls.create_pg_interfaces([0])
|
|
cls.pg0.config_ip4()
|
|
cls.pg0.generate_remote_hosts()
|
|
cls.pg0.configure_ipv4_neighbors()
|
|
cls.pg0.admin_up()
|
|
cls.pg0.resolve_arp()
|
|
|
|
except Exception:
|
|
super(BFDTestCase, cls).tearDownClass()
|
|
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()
|
|
self.vpp_session.admin_up()
|
|
self.test_session = BFDTestSession(self, self.pg0)
|
|
|
|
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("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,
|
|
"BFD interface index")
|
|
is_ipv6 = 0
|
|
if self.vpp_session.af == AF_INET6:
|
|
is_ipv6 = 1
|
|
self.assert_equal(e.is_ipv6, is_ipv6, "is_ipv6")
|
|
if self.vpp_session.af == AF_INET:
|
|
self.assert_equal(e.local_addr[:4], self.vpp_session.local_addr_n,
|
|
"Local IPv4 address")
|
|
self.assert_equal(e.peer_addr[:4], self.vpp_session.peer_addr_n,
|
|
"Peer IPv4 address")
|
|
else:
|
|
self.assert_equal(e.local_addr, self.vpp_session.local_addr_n,
|
|
"Local IPv6 address")
|
|
self.assert_equal(e.peer_addr, self.vpp_session.peer_addr_n,
|
|
"Peer IPv6 address")
|
|
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:
|
|
raise Exception(ppp("Unexpected or invalid BFD packet:", p))
|
|
if bfd.payload:
|
|
raise Exception(ppp("Unexpected payload in BFD packet:", bfd))
|
|
verify_ip(self, p, self.pg0.local_ip4, self.pg0.remote_ip4)
|
|
verify_udp(self, p)
|
|
self.test_session.verify_packet(p)
|
|
return p
|
|
|
|
def test_slow_timer(self):
|
|
""" verify slow periodic control frames while session down """
|
|
self.pg_enable_capture([self.pg0])
|
|
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(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.70, 1.05, "time between slow packets")
|
|
before = after
|
|
|
|
def test_zero_remote_min_rx(self):
|
|
""" no packets when zero BFD RemoteMinRxInterval """
|
|
self.pg_enable_capture([self.pg0])
|
|
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,
|
|
required_min_rx_interval=0)
|
|
self.test_session.send_packet()
|
|
e = self.vapi.wait_for_event(1, "bfd_udp_session_details")
|
|
self.verify_event(e, expected_state=BFDState.up)
|
|
|
|
try:
|
|
p = self.pg0.wait_for_packet(timeout=1)
|
|
except:
|
|
return
|
|
raise Exception(ppp("Received unexpected BFD packet:", p))
|
|
|
|
def bfd_session_up(self):
|
|
self.pg_enable_capture([self.pg0])
|
|
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=100000)
|
|
self.test_session.send_packet()
|
|
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("BFD: Session is Up")
|
|
self.test_session.update(state=BFDState.up)
|
|
|
|
def test_session_up(self):
|
|
""" bring BFD session up """
|
|
self.bfd_session_up()
|
|
|
|
def test_hold_up(self):
|
|
""" 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):
|
|
""" verify session goes down after inactivity """
|
|
self.bfd_session_up()
|
|
self.wait_for_bfd_packet()
|
|
self.assert_equal(len(self.vapi.collect_events()), 0,
|
|
"number of bfd events")
|
|
self.wait_for_bfd_packet()
|
|
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)
|
|
|
|
def test_large_required_min_rx(self):
|
|
""" 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 = 0
|
|
while time.time() < now + interval / 1000000:
|
|
try:
|
|
p = self.wait_for_bfd_packet()
|
|
if count > 1:
|
|
self.logger.error(ppp("Received unexpected packet:", p))
|
|
count += 1
|
|
except:
|
|
pass
|
|
self.assert_in_range(count, 0, 1, "number of packets received")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(testRunner=VppTestRunner)
|