X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_bfd.py;h=d047b5a3b99da23615c6017082ee5d715938b368;hb=50132fa8c14dd74a77b760bc603f96d7027bb73a;hp=1ea69f5507b43b8b1401809cea64cbf782bffe30;hpb=dab231a11ec96e829b22ff80c612333edc5a93e6;p=vpp.git diff --git a/test/test_bfd.py b/test/test_bfd.py index 1ea69f5507b..d047b5a3b99 100644 --- a/test/test_bfd.py +++ b/test/test_bfd.py @@ -7,6 +7,8 @@ from bfd import * from framework import * from util import ppp +us_in_sec = 1000000 + class BFDAPITestCase(VppTestCase): """Bidirectional Forwarding Detection (BFD) - API""" @@ -16,9 +18,10 @@ class BFDAPITestCase(VppTestCase): super(BFDAPITestCase, cls).setUpClass() try: - cls.create_pg_interfaces([0]) - cls.pg0.config_ip4() - cls.pg0.resolve_arp() + cls.create_pg_interfaces(range(2)) + for i in cls.pg_interfaces: + i.config_ip4() + i.resolve_arp() except Exception: super(BFDAPITestCase, cls).tearDownClass() @@ -39,45 +42,31 @@ class BFDAPITestCase(VppTestCase): """ 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") + with self.vapi.expect_negative_api_retval(): + session.add_vpp_config() -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") - + session.remove_vpp_config() -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") + def test_add_two(self): + """ create two BFD sessions """ + session1 = VppBFDUDPSession(self, self.pg0, self.pg0.remote_ip4) + session1.add_vpp_config() + session2 = VppBFDUDPSession(self, self.pg1, self.pg1.remote_ip4) + session2.add_vpp_config() + self.assertNotEqual(session1.bs_index, session2.bs_index, + "Different BFD sessions share bs_index (%s)" % + session1.bs_index) class BFDTestSession(object): + """ BFD session as seen from test framework side """ - def __init__(self, test, interface, detect_mult=3): + def __init__(self, test, interface, af, detect_mult=3): self.test = test + self.af = af self.interface = interface + self.udp_sport = 50000 self.bfd_values = { 'my_discriminator': 0, 'desired_min_tx_interval': 100000, @@ -89,8 +78,25 @@ class BFDTestSession(object): self.bfd_values.update(kwargs) def create_packet(self): - packet = create_packet(self.interface) + if self.af == AF_INET6: + packet = (Ether(src=self.interface.remote_mac, + dst=self.interface.local_mac) / + IPv6(src=self.interface.remote_ip6, + dst=self.interface.local_ip6, + hlim=255) / + UDP(sport=self.udp_sport, dport=BFD.udp_dport) / + BFD()) + else: + packet = (Ether(src=self.interface.remote_mac, + dst=self.interface.local_mac) / + IP(src=self.interface.remote_ip4, + dst=self.interface.local_ip4, + ttl=255) / + UDP(sport=self.udp_sport, dport=BFD.udp_dport) / + BFD()) + self.test.logger.debug("BFD: Creating packet") for name, value in self.bfd_values.iteritems(): + self.test.logger.debug("BFD: setting packet.%s=%s", name, value) packet[BFD].setfieldval(name, value) return packet @@ -109,38 +115,52 @@ class BFDTestSession(object): "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) +class BFDCommonCode: + """Common code used by both IPv4 and IPv6 Test Cases""" 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.vapi.want_bfd_events(enable_disable=0) self.vpp_session.remove_vpp_config() - super(BFDTestCase, self).tearDown() + + def bfd_session_up(self): + self.pg_enable_capture([self.pg0]) + self.logger.info("BFD: Waiting for slow hello") + p, timeout = 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 verify_ip(self, packet): + """ Verify correctness of IP layer. """ + if self.vpp_session.af == AF_INET6: + ip = packet[IPv6] + local_ip = self.pg0.local_ip6 + remote_ip = self.pg0.remote_ip6 + self.assert_equal(ip.hlim, 255, "IPv6 hop limit") + else: + ip = packet[IP] + local_ip = self.pg0.local_ip4 + remote_ip = self.pg0.remote_ip4 + self.assert_equal(ip.ttl, 255, "IPv4 TTL") + self.assert_equal(ip.src, local_ip, "IP source address") + self.assert_equal(ip.dst, remote_ip, "IP destination address") + + def verify_udp(self, packet): + """ Verify correctness of UDP layer. """ + udp = packet[UDP] + self.assert_equal(udp.dport, BFD.udp_dport, "UDP destination port") + self.assert_in_range(udp.sport, BFD.udp_sport_min, BFD.udp_sport_max, + "UDP source port") def verify_event(self, event, expected_state): """ Verify correctness of event values. """ @@ -148,8 +168,10 @@ class BFDTestCase(VppTestCase): 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") + 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 @@ -167,17 +189,73 @@ class BFDTestCase(VppTestCase): self.assert_equal(e.state, expected_state, BFDState) def wait_for_bfd_packet(self, timeout=1): + """ wait for BFD packet + + :param timeout: how long to wait max + + :returns: tuple (packet, time spent waiting for packet) + """ self.logger.info("BFD: Waiting for BFD packet") + before = time.time() p = self.pg0.wait_for_packet(timeout=timeout) + after = time.time() + self.logger.debug(ppp("Got packet:", p)) 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.verify_ip(p) + self.verify_udp(p) self.test_session.verify_packet(p) - return p + return p, after - before + + 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() + + +class BFD4TestCase(VppTestCase, BFDCommonCode): + """Bidirectional Forwarding Detection (BFD)""" + + @classmethod + def setUpClass(cls): + super(BFD4TestCase, 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(BFD4TestCase, cls).tearDownClass() + raise + + def setUp(self): + super(BFD4TestCase, self).setUp() + self.vapi.want_bfd_events() + try: + 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, AF_INET) + except: + self.vapi.want_bfd_events(enable_disable=0) + raise + + def tearDown(self): + BFDCommonCode.tearDown(self) + super(BFD4TestCase, self).tearDown() def test_slow_timer(self): """ verify slow periodic control frames while session down """ @@ -198,7 +276,7 @@ class BFDTestCase(VppTestCase): 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) + p, timeout = self.wait_for_bfd_packet(2) self.test_session.update(my_discriminator=randint(0, 40000000), your_discriminator=p[BFD].my_discriminator, state=BFDState.init, @@ -213,33 +291,6 @@ class BFDTestCase(VppTestCase): 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() @@ -260,7 +311,7 @@ class BFDTestCase(VppTestCase): self.test_session.send_packet() now = time.time() count = 0 - while time.time() < now + interval / 1000000: + while time.time() < now + interval / us_in_sec: try: p = self.wait_for_bfd_packet() if count > 1: @@ -270,6 +321,68 @@ class BFDTestCase(VppTestCase): pass self.assert_in_range(count, 0, 1, "number of packets received") + def test_immediate_remote_min_rx_reduce(self): + """ immediately honor remote min rx reduction """ + self.vpp_session.remove_vpp_config() + self.vpp_session = VppBFDUDPSession( + self, self.pg0, self.pg0.remote_ip4, desired_min_tx=10000) + self.vpp_session.add_vpp_config() + self.test_session.update(desired_min_tx_interval=1000000, + required_min_rx_interval=1000000) + self.bfd_session_up() + self.wait_for_bfd_packet() + interval = 100000 + self.test_session.update(required_min_rx_interval=interval) + self.test_session.send_packet() + p, ttp = self.wait_for_bfd_packet() + # allow extra 10% to work around timing issues, first packet is special + self.assert_in_range(ttp, 0, 1.10 * interval / us_in_sec, + "time between BFD packets") + p, ttp = self.wait_for_bfd_packet() + self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec, + 1.10 * interval / us_in_sec, + "time between BFD packets") + p, ttp = self.wait_for_bfd_packet() + self.assert_in_range(ttp, .9 * .75 * interval / us_in_sec, + 1.10 * interval / us_in_sec, + "time between BFD packets") + + +class BFD6TestCase(VppTestCase, BFDCommonCode): + """Bidirectional Forwarding Detection (BFD) (IPv6) """ + + @classmethod + def setUpClass(cls): + super(BFD6TestCase, cls).setUpClass() + try: + cls.create_pg_interfaces([0]) + cls.pg0.config_ip6() + cls.pg0.configure_ipv6_neighbors() + cls.pg0.admin_up() + cls.pg0.resolve_ndp() + + except Exception: + super(BFD6TestCase, cls).tearDownClass() + raise + + def setUp(self): + super(BFD6TestCase, self).setUp() + self.vapi.want_bfd_events() + try: + self.vpp_session = VppBFDUDPSession(self, self.pg0, + self.pg0.remote_ip6, + af=AF_INET6) + self.vpp_session.add_vpp_config() + self.vpp_session.admin_up() + self.test_session = BFDTestSession(self, self.pg0, AF_INET6) + self.logger.debug(self.vapi.cli("show adj nbr")) + except: + self.vapi.want_bfd_events(enable_disable=0) + raise + + def tearDown(self): + BFDCommonCode.tearDown(self) + super(BFD6TestCase, self).tearDown() if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)