make_test: add icmp packet size sweep and icmp echo tests

Jira: CSIT-1141

Change-Id: I162bb4e718bff188abefc7b2f33501de9c55bb03
Signed-off-by: Jan Gelety <jgelety@cisco.com>

1 files changed, 101 insertions, 23 deletions

diff --git a/test/test_ip6.py b/test/test_ip6.py
index 21d8025d583..2f19bcf1547 100644
--- a/test/test_ip6.py
+++ b/test/test_ip6.py
@@ -18,7 +18,7 @@ from scapy.layers.inet6 import IPv6, UDP, TCP, ICMPv6ND_NS, ICMPv6ND_RS, \
     ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation, \
     ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
     ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, \
-    ICMPv6TimeExceeded
+    ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply
 from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
     in6_mactoifaceid, in6_ismaddr
 from scapy.utils import inet_pton, inet_ntop
@@ -172,7 +172,6 @@ class TestIPv6(TestIPv6ND):
 
         :ivar list interfaces: pg interfaces and subinterfaces.
         :ivar dict flows: IPv4 packet flows in test.
-        :ivar list pg_if_packet_sizes: packet sizes in test.
 
         *TODO:* Create AD sub interface
         """
@@ -195,8 +194,7 @@ class TestIPv6(TestIPv6ND):
         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
 
         # packet sizes
-        self.pg_if_packet_sizes = [64, 512, 1518, 9018]
-        self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
+        self.pg_if_packet_sizes = [64, 1500, 9020]
 
         self.interfaces = list(self.pg_interfaces)
         self.interfaces.extend(self.sub_interfaces)
@@ -250,27 +248,45 @@ class TestIPv6(TestIPv6ND):
                                      (count, percent))
                     percent += 1
 
-    def create_stream(self, src_if, packet_sizes):
+    def modify_packet(self, src_if, packet_size, pkt):
+        """Add load, set destination IP and extend packet to required packet
+        size for defined interface.
+
+        :param VppInterface src_if: Interface to create packet for.
+        :param int packet_size: Required packet size.
+        :param Scapy pkt: Packet to be modified.
+        """
+        dst_if_idx = packet_size / 10 % 2
+        dst_if = self.flows[src_if][dst_if_idx]
+        info = self.create_packet_info(src_if, dst_if)
+        payload = self.info_to_payload(info)
+        p = pkt/Raw(payload)
+        p[IPv6].dst = dst_if.remote_ip6
+        info.data = p.copy()
+        if isinstance(src_if, VppSubInterface):
+            p = src_if.add_dot1_layer(p)
+        self.extend_packet(p, packet_size)
+
+        return p
+
+    def create_stream(self, src_if):
         """Create input packet stream for defined interface.
 
         :param VppInterface src_if: Interface to create packet stream for.
-        :param list packet_sizes: Required packet sizes.
         """
-        pkts = []
-        for i in range(0, 257):
-            dst_if = self.flows[src_if][i % 2]
-            info = self.create_packet_info(src_if, dst_if)
-            payload = self.info_to_payload(info)
-            p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
-                 IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
-                 UDP(sport=1234, dport=1234) /
-                 Raw(payload))
-            info.data = p.copy()
-            if isinstance(src_if, VppSubInterface):
-                p = src_if.add_dot1_layer(p)
-            size = packet_sizes[(i // 2) % len(packet_sizes)]
-            self.extend_packet(p, size)
-            pkts.append(p)
+        hdr_ext = 4 if isinstance(src_if, VppSubInterface) else 0
+        pkt_tmpl = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
+                    IPv6(src=src_if.remote_ip6) /
+                    UDP(sport=1234, dport=1234))
+
+        pkts = [self.modify_packet(src_if, i, pkt_tmpl)
+                for i in xrange(self.pg_if_packet_sizes[0],
+                                self.pg_if_packet_sizes[1], 10)]
+        pkts_b = [self.modify_packet(src_if, i, pkt_tmpl)
+                  for i in xrange(self.pg_if_packet_sizes[1] + hdr_ext,
+                                  self.pg_if_packet_sizes[2] + hdr_ext, 50)]
+        pkts.extend(pkts_b)
+
         return pkts
 
     def verify_capture(self, dst_if, capture):
@@ -333,11 +349,11 @@ class TestIPv6(TestIPv6ND):
             - Send and verify received packets on each interface.
         """
 
-        pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
+        pkts = self.create_stream(self.pg0)
         self.pg0.add_stream(pkts)
 
         for i in self.sub_interfaces:
-            pkts = self.create_stream(i, self.sub_if_packet_sizes)
+            pkts = self.create_stream(i)
             i.parent.add_stream(pkts)
 
         self.pg_enable_capture(self.pg_interfaces)
@@ -898,6 +914,68 @@ class TestIPv6(TestIPv6ND):
         self.pg0.ip6_ra_config(no=1, suppress=1, send_unicast=0)
 
 
+class TestICMPv6Echo(VppTestCase):
+    """ ICMPv6 Echo Test Case """
+
+    def setUp(self):
+        super(TestICMPv6Echo, self).setUp()
+
+        # create 1 pg interface
+        self.create_pg_interfaces(range(1))
+
+        for i in self.pg_interfaces:
+            i.admin_up()
+            i.config_ip6()
+            i.resolve_ndp()
+
+    def tearDown(self):
+        super(TestICMPv6Echo, self).tearDown()
+        for i in self.pg_interfaces:
+            i.unconfig_ip6()
+            i.ip6_disable()
+            i.admin_down()
+
+    def test_icmpv6_echo(self):
+        """ VPP replies to ICMPv6 Echo Request
+
+        Test scenario:
+
+            - Receive ICMPv6 Echo Request message on pg0 interface.
+            - Check outgoing ICMPv6 Echo Reply message on pg0 interface.
+        """
+
+        icmpv6_id = 0xb
+        icmpv6_seq = 5
+        icmpv6_data = '\x0a' * 18
+        p_echo_request = (Ether(src=self.pg0.remote_mac,
+                                dst=self.pg0.local_mac) /
+                          IPv6(src=self.pg0.remote_ip6,
+                               dst=self.pg0.local_ip6) /
+                          ICMPv6EchoRequest(id=icmpv6_id, seq=icmpv6_seq,
+                                            data=icmpv6_data))
+
+        self.pg0.add_stream(p_echo_request)
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pg_start()
+
+        rx = self.pg0.get_capture(1)
+        rx = rx[0]
+        ether = rx[Ether]
+        ipv6 = rx[IPv6]
+        icmpv6 = rx[ICMPv6EchoReply]
+
+        self.assertEqual(ether.src, self.pg0.local_mac)
+        self.assertEqual(ether.dst, self.pg0.remote_mac)
+
+        self.assertEqual(ipv6.src, self.pg0.local_ip6)
+        self.assertEqual(ipv6.dst, self.pg0.remote_ip6)
+
+        self.assertEqual(icmp6types[icmpv6.type], "Echo Reply")
+        self.assertEqual(icmpv6.id, icmpv6_id)
+        self.assertEqual(icmpv6.seq, icmpv6_seq)
+        self.assertEqual(icmpv6.data, icmpv6_data)
+
+
 class TestIPv6RD(TestIPv6ND):
     """ IPv6 Router Discovery Test Case """