aboutsummaryrefslogtreecommitdiffstats
path: root/test/test_ip4.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_ip4.py')
-rw-r--r--test/test_ip4.py124
1 files changed, 100 insertions, 24 deletions
diff --git a/test/test_ip4.py b/test/test_ip4.py
index 83129e2462d..2d98ed5297b 100644
--- a/test/test_ip4.py
+++ b/test/test_ip4.py
@@ -36,7 +36,6 @@ class TestIPv4(VppTestCase):
: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.
"""
super(TestIPv4, self).setUp()
@@ -55,8 +54,7 @@ class TestIPv4(VppTestCase):
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)
@@ -103,34 +101,52 @@ class TestIPv4(VppTestCase):
(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[IP].dst = dst_if.remote_ip4
+ 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) /
- IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
- 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) /
+ IP(src=src_if.remote_ip4) /
+ 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):
"""Verify captured input packet stream for defined interface.
:param VppInterface dst_if: Interface to verify captured packet stream
- for.
+ for.
:param list capture: Captured packet stream.
"""
self.logger.info("Verifying capture on interface %s" % dst_if.name)
@@ -183,15 +199,15 @@ class TestIPv4(VppTestCase):
Test scenario:
- Create IPv4 stream for pg0 interface
- - Create IPv4 tagged streams for pg1's and pg2's subinterface.
+ - Create IPv4 tagged streams for pg1's and pg2's sub-interface.
- 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)
@@ -205,6 +221,66 @@ class TestIPv4(VppTestCase):
self.verify_capture(i, pkts)
+class TestICMPEcho(VppTestCase):
+ """ ICMP Echo Test Case """
+
+ def setUp(self):
+ super(TestICMPEcho, self).setUp()
+
+ # create 1 pg interface
+ self.create_pg_interfaces(range(1))
+
+ for i in self.pg_interfaces:
+ i.admin_up()
+ i.config_ip4()
+ i.resolve_arp()
+
+ def tearDown(self):
+ super(TestICMPEcho, self).tearDown()
+ for i in self.pg_interfaces:
+ i.unconfig_ip4()
+ i.admin_down()
+
+ def test_icmp_echo(self):
+ """ VPP replies to ICMP Echo Request
+
+ Test scenario:
+
+ - Receive ICMP Echo Request message on pg0 interface.
+ - Check outgoing ICMP Echo Reply message on pg0 interface.
+ """
+
+ icmp_id = 0xb
+ icmp_seq = 5
+ icmp_load = '\x0a' * 18
+ p_echo_request = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
+ ICMP(id=icmp_id, seq=icmp_seq) /
+ Raw(load=icmp_load))
+
+ 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]
+ ipv4 = rx[IP]
+ icmp = rx[ICMP]
+
+ self.assertEqual(ether.src, self.pg0.local_mac)
+ self.assertEqual(ether.dst, self.pg0.remote_mac)
+
+ self.assertEqual(ipv4.src, self.pg0.local_ip4)
+ self.assertEqual(ipv4.dst, self.pg0.remote_ip4)
+
+ self.assertEqual(icmptypes[icmp.type], "echo-reply")
+ self.assertEqual(icmp.id, icmp_id)
+ self.assertEqual(icmp.seq, icmp_seq)
+ self.assertEqual(icmp[Raw].load, icmp_load)
+
+
class TestIPv4FibCrud(VppTestCase):
""" FIB - add/update/delete - ip4 routes