X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_srv6.py;h=449ad59ac60ddc60171e0294b804641c17a92f9e;hb=4c45d6fef426863cce7a6f6cb42150897a674ed9;hp=cc53c6b1e79ee20294752f16f0d6cd4cf4f117b4;hpb=a7427ec6f86cbeba7594f98e41fecab291d66b73;p=vpp.git diff --git a/test/test_srv6.py b/test/test_srv6.py index cc53c6b1e79..449ad59ac60 100644 --- a/test/test_srv6.py +++ b/test/test_srv6.py @@ -1,11 +1,11 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 import unittest import binascii from socket import AF_INET6 from framework import VppTestCase, VppTestRunner -from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto, VppIpTable +from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto, VppIpTable from vpp_srv6 import SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, \ SRv6PolicyType, VppSRv6Steering, SRv6PolicySteeringTypes @@ -15,8 +15,6 @@ from scapy.layers.l2 import Ether, Dot1Q from scapy.layers.inet6 import IPv6, UDP, IPv6ExtHdrSegmentRouting from scapy.layers.inet import IP, UDP -from scapy.utils import inet_pton, inet_ntop - from util import ppp @@ -24,8 +22,12 @@ class TestSRv6(VppTestCase): """ SRv6 Test Case """ @classmethod - def setUpClass(self): - super(TestSRv6, self).setUpClass() + def setUpClass(cls): + super(TestSRv6, cls).setUpClass() + + @classmethod + def tearDownClass(cls): + super(TestSRv6, cls).tearDownClass() def setUp(self): """ Perform test setup before each test case. @@ -112,7 +114,7 @@ class TestSRv6(VppTestCase): if any(ipv6): self.logger.debug(self.vapi.cli("show ip6 neighbors")) if any(ipv4): - self.logger.debug(self.vapi.cli("show ip arp")) + self.logger.debug(self.vapi.cli("show ip4 neighbors")) self.logger.debug(self.vapi.cli("show interface")) self.logger.debug(self.vapi.cli("show hardware")) @@ -142,9 +144,7 @@ class TestSRv6(VppTestCase): # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, - self.pg1.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + self.pg1.sw_if_index)]) route.add_vpp_config() # configure encaps IPv6 source address @@ -244,9 +244,7 @@ class TestSRv6(VppTestCase): # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, - self.pg1.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + self.pg1.sw_if_index)]) route.add_vpp_config() # configure encaps IPv6 source address @@ -337,9 +335,7 @@ class TestSRv6(VppTestCase): # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, - self.pg1.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + self.pg1.sw_if_index)]) route.add_vpp_config() # configure encaps IPv6 source address @@ -421,9 +417,7 @@ class TestSRv6(VppTestCase): # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, - self.pg1.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + self.pg1.sw_if_index)]) route.add_vpp_config() # configure encaps IPv6 source address @@ -508,17 +502,14 @@ class TestSRv6(VppTestCase): # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, - self.pg1.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + self.pg1.sw_if_index)]) route.add_vpp_config() # configure SRv6 localSID End without PSP behavior localsid = VppSRv6LocalSID( - self, localsid={'addr': 'A3::0'}, + self, localsid='A3::0', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_END, - nh_addr4='0.0.0.0', - nh_addr6='::', + nh_addr=0, end_psp=0, sw_if_index=0, vlan_index=0, @@ -554,9 +545,18 @@ class TestSRv6(VppTestCase): # TODO: test behavior with SL=0 packet (needs 2*SRH?) + expected_count = len(pkts) + + # packets without SRH (should not crash) + packet_header = self.create_packet_header_IPv6('a3::') + # create traffic stream pg0->pg1 + pkts.extend(self.create_stream(self.pg0, self.pg1, packet_header, + self.pg_packet_sizes, count)) + # send packets and verify received packets self.send_and_verify_pkts(self.pg0, pkts, self.pg1, - self.compare_rx_tx_packet_End) + self.compare_rx_tx_packet_End, + expected_count=expected_count) # log the localsid counters self.logger.info(self.vapi.cli("show sr localsid")) @@ -580,17 +580,14 @@ class TestSRv6(VppTestCase): # configure FIB entries route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, - self.pg1.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + self.pg1.sw_if_index)]) route.add_vpp_config() # configure SRv6 localSID End with PSP behavior localsid = VppSRv6LocalSID( - self, localsid={'addr': 'A3::0'}, + self, localsid='A3::0', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_END, - nh_addr4='0.0.0.0', - nh_addr6='::', + nh_addr=0, end_psp=1, sw_if_index=0, vlan_index=0, @@ -651,22 +648,18 @@ class TestSRv6(VppTestCase): # a4::/64 via pg1 and pg2 route = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, - self.pg1.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6), + self.pg1.sw_if_index), VppRoutePath(self.pg2.remote_ip6, - self.pg2.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + self.pg2.sw_if_index)]) route.add_vpp_config() self.logger.debug(self.vapi.cli("show ip6 fib")) # configure SRv6 localSID End.X without PSP behavior # End.X points to interface pg1 localsid = VppSRv6LocalSID( - self, localsid={'addr': 'A3::C4'}, + self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_X, - nh_addr4='0.0.0.0', - nh_addr6=self.pg1.remote_ip6, + nh_addr=self.pg1.remote_ip6, end_psp=0, sw_if_index=self.pg1.sw_if_index, vlan_index=0, @@ -730,21 +723,18 @@ class TestSRv6(VppTestCase): # configure FIB entries # a4::/64 via pg1 and pg2 route = VppIpRoute(self, "a4::", 64, - [VppRoutePath(self.pg1.remote_ip6, - self.pg1.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6), + [VppRoutePath( + self.pg1.remote_ip6, + self.pg1.sw_if_index), VppRoutePath(self.pg2.remote_ip6, - self.pg2.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + self.pg2.sw_if_index)]) route.add_vpp_config() # configure SRv6 localSID End with PSP behavior localsid = VppSRv6LocalSID( - self, localsid={'addr': 'A3::C4'}, + self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_X, - nh_addr4='0.0.0.0', - nh_addr6=self.pg1.remote_ip6, + nh_addr=self.pg1.remote_ip6, end_psp=1, sw_if_index=self.pg1.sw_if_index, vlan_index=0, @@ -807,10 +797,9 @@ class TestSRv6(VppTestCase): # configure SRv6 localSID End.DX6 behavior localsid = VppSRv6LocalSID( - self, localsid={'addr': 'A3::C4'}, + self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DX6, - nh_addr4='0.0.0.0', - nh_addr6=self.pg1.remote_ip6, + nh_addr=self.pg1.remote_ip6, end_psp=0, sw_if_index=self.pg1.sw_if_index, vlan_index=0, @@ -876,18 +865,14 @@ class TestSRv6(VppTestCase): route0 = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6, nh_table_id=0)], - table_id=0, - is_ip6=1) + table_id=0) route0.add_vpp_config() route1 = VppIpRoute(self, "a4::", 64, [VppRoutePath(self.pg2.remote_ip6, self.pg2.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6, nh_table_id=vrf_1)], - table_id=vrf_1, - is_ip6=1) + table_id=vrf_1) route1.add_vpp_config() self.logger.debug(self.vapi.cli("show ip6 fib")) @@ -896,10 +881,9 @@ class TestSRv6(VppTestCase): # fib_table: where the localsid is installed # sw_if_index: in T-variants of localsid this is the vrf table_id localsid = VppSRv6LocalSID( - self, localsid={'addr': 'A3::C4'}, + self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DT6, - nh_addr4='0.0.0.0', - nh_addr6='::', + nh_addr=0, end_psp=0, sw_if_index=vrf_1, vlan_index=0, @@ -962,10 +946,9 @@ class TestSRv6(VppTestCase): # configure SRv6 localSID End.DX4 behavior localsid = VppSRv6LocalSID( - self, localsid={'addr': 'A3::C4'}, + self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DX4, - nh_addr4=self.pg1.remote_ip4, - nh_addr6='::', + nh_addr=self.pg1.remote_ip4, end_psp=0, sw_if_index=self.pg1.sw_if_index, vlan_index=0, @@ -1034,15 +1017,13 @@ class TestSRv6(VppTestCase): [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index, nh_table_id=0)], - table_id=0, - is_ip6=0) + table_id=0) route0.add_vpp_config() route1 = VppIpRoute(self, "4.1.1.0", 24, [VppRoutePath(self.pg2.remote_ip4, self.pg2.sw_if_index, nh_table_id=vrf_1)], - table_id=vrf_1, - is_ip6=0) + table_id=vrf_1) route1.add_vpp_config() self.logger.debug(self.vapi.cli("show ip fib")) @@ -1051,10 +1032,9 @@ class TestSRv6(VppTestCase): # fib_table: where the localsid is installed # sw_if_index: in T-variants of localsid: vrf table_id localsid = VppSRv6LocalSID( - self, localsid={'addr': 'A3::C4'}, + self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DT4, - nh_addr4='0.0.0.0', - nh_addr6='::', + nh_addr=0, end_psp=0, sw_if_index=vrf_1, vlan_index=0, @@ -1116,10 +1096,9 @@ class TestSRv6(VppTestCase): # configure SRv6 localSID End.DX2 behavior localsid = VppSRv6LocalSID( - self, localsid={'addr': 'A3::C4'}, + self, localsid='A3::C4', behavior=SRv6LocalSIDBehaviors.SR_BEHAVIOR_DX2, - nh_addr4='0.0.0.0', - nh_addr6='::', + nh_addr=0, end_psp=0, sw_if_index=self.pg1.sw_if_index, vlan_index=0, @@ -1196,10 +1175,9 @@ class TestSRv6(VppTestCase): # configure FIB entries route = VppIpRoute(self, "a4::", 64, - [VppRoutePath(self.pg4.remote_ip6, - self.pg4.sw_if_index, - proto=DpoProto.DPO_PROTO_IP6)], - is_ip6=1) + [VppRoutePath( + self.pg4.remote_ip6, + self.pg4.sw_if_index)]) route.add_vpp_config() # configure encaps IPv6 source address @@ -1235,7 +1213,7 @@ class TestSRv6(VppTestCase): self.assertIsNotNone(r, 'No response msg for add_del_table') table_index = r.new_table_index - # add the source routign node as a ip6 inacl netxt node + # add the source routing node as a ip6 inacl netxt node r = self.vapi.add_node_next('ip6-inacl', 'sr-pl-rewrite-insert') inacl_next_node_index = r.node_index @@ -1486,8 +1464,8 @@ class TestSRv6(VppTestCase): self.assertEqual(rx_srh.segleft, len(tx_seglist)-1) # segleft should be equal to lastentry self.assertEqual(rx_srh.segleft, rx_srh.lastentry) - # nh should be "No Next Header" (59) - self.assertEqual(rx_srh.nh, 59) + # nh should be "No Next Header" (143) + self.assertEqual(rx_srh.nh, 143) # the whole rx'ed pkt beyond SRH should be equal to tx'ed pkt self.assertEqual(Ether(scapy.compat.raw(rx_srh.payload)), tx_ether) @@ -1845,13 +1823,16 @@ class TestSRv6(VppTestCase): self.logger.info("Done creating packets") return pkts - def send_and_verify_pkts(self, input, pkts, output, compare_func): + def send_and_verify_pkts(self, input, pkts, output, compare_func, + expected_count=None): """Send packets and verify received packets using compare_func :param input: ingress interface of DUT :param pkts: list of packets to transmit :param output: egress interface of DUT :param compare_func: function to compare in and out packets + :param expected_count: expected number of captured packets (if + different than len(pkts)) """ # add traffic stream to input interface input.add_stream(pkts) @@ -1865,7 +1846,7 @@ class TestSRv6(VppTestCase): # get output capture self.logger.info("Getting packet capture") - capture = output.get_capture() + capture = output.get_capture(expected_count=expected_count) # assert nothing was captured on input interface input.assert_nothing_captured() @@ -2052,7 +2033,7 @@ class TestSRv6(VppTestCase): p = (IPv6(src='1234::1', dst=sidlist[segleft]) / IPv6ExtHdrSegmentRouting(addresses=sidlist, - segleft=segleft, nh=59) / + segleft=segleft, nh=143) / eth) return p @@ -2071,7 +2052,7 @@ class TestSRv6(VppTestCase): else: eth.type = etype - p = (IPv6(src='1234::1', dst=dst_outer, nh=59) / eth) + p = (IPv6(src='1234::1', dst=dst_outer, nh=143) / eth) return p def get_payload_info(self, packet): @@ -2148,13 +2129,18 @@ class TestSRv6(VppTestCase): self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise + # FIXME: there is no need to check manually that all the packets + # arrived (already done so by get_capture); checking here + # prevents testing packets that are expected to be dropped, so + # commenting this out for now + # have all expected packets arrived? - for i in self.pg_interfaces: - remaining_packet = self.get_next_packet_info_for_interface2( - i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index]) - self.assertTrue(remaining_packet is None, - "Interface %s: Packet expected from interface %s " - "didn't arrive" % (dst_if.name, i.name)) + # for i in self.pg_interfaces: + # remaining_packet = self.get_next_packet_info_for_interface2( + # i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index]) + # self.assertTrue(remaining_packet is None, + # "Interface %s: Packet expected from interface %s " + # "didn't arrive" % (dst_if.name, i.name)) if __name__ == '__main__':