test/test_ping.py

   1 import socket
   2
   3 from scapy.layers.inet import IP, UDP, ICMP
   4 from scapy.layers.inet6 import IPv6
   5 from scapy.layers.l2 import Ether, GRE
   6 from scapy.packet import Raw
   7
   8 from framework import VppTestCase
   9 from util import ppp
  10
  11 """ TestPing is a subclass of  VPPTestCase classes.
  12
  13 Basic test for sanity check of the ping.
  14
  15 """
  16
  17
  18 class TestPing(VppTestCase):
  19     """ Ping Test Case """
  20
  21     @classmethod
  22     def setUpClass(cls):
  23         super(TestPing, cls).setUpClass()
  24         try:
  25             cls.create_pg_interfaces(range(2))
  26             cls.interfaces = list(cls.pg_interfaces)
  27
  28             for i in cls.interfaces:
  29                 i.admin_up()
  30                 i.config_ip4()
  31                 i.config_ip6()
  32                 i.disable_ipv6_ra()
  33                 i.resolve_arp()
  34                 i.resolve_ndp()
  35         except Exception:
  36             super(TestPing, cls).tearDownClass()
  37             raise
  38
  39     def tearDown(self):
  40         super(TestPing, self).tearDown()
  41         if not self.vpp_dead:
  42             self.vapi.cli("show hardware")
  43
  44     def test_ping_basic(self):
  45         """ basic ping test """
  46         try:
  47             self.pg_enable_capture(self.pg_interfaces)
  48             self.pg_start()
  49             self.logger.info(self.vapi.cli("show ip arp"))
  50             self.logger.info(self.vapi.cli("show ip6 neighbors"))
  51
  52             remote_ip4 = self.pg1.remote_ip4
  53             ping_cmd = "ping " + remote_ip4 + " interval 0.01 repeat 10"
  54             ret = self.vapi.cli(ping_cmd)
  55             self.logger.info(ret)
  56             out = self.pg1.get_capture(10)
  57             icmp_id = None
  58             icmp_seq = 1
  59             for p in out:
  60                 ip = p[IP]
  61                 self.assertEqual(ip.version, 4)
  62                 self.assertEqual(ip.flags, 0)
  63                 self.assertEqual(ip.src, self.pg1.local_ip4)
  64                 self.assertEqual(ip.dst, self.pg1.remote_ip4)
  65                 self.assertEqual(ip.proto, 1)
  66                 self.assertEqual(len(ip.options), 0)
  67                 self.assertGreaterEqual(ip.ttl, 254)
  68                 icmp = p[ICMP]
  69                 self.assertEqual(icmp.type, 8)
  70                 self.assertEqual(icmp.code, 0)
  71                 self.assertEqual(icmp.seq, icmp_seq)
  72                 icmp_seq = icmp_seq + 1
  73                 if icmp_id is None:
  74                     icmp_id = icmp.id
  75                 else:
  76                     self.assertEqual(icmp.id, icmp_id)
  77         finally:
  78             self.vapi.cli("show error")
  79
  80     def test_ping_burst(self):
  81         """ burst ping test """
  82         try:
  83             self.pg_enable_capture(self.pg_interfaces)
  84             self.pg_start()
  85             self.logger.info(self.vapi.cli("show ip arp"))
  86             self.logger.info(self.vapi.cli("show ip6 neighbors"))
  87
  88             remote_ip4 = self.pg1.remote_ip4
  89             ping_cmd = "ping " + remote_ip4 + " interval 0.01 burst 3"
  90             ret = self.vapi.cli(ping_cmd)
  91             self.logger.info(ret)
  92             out = self.pg1.get_capture(3*5)
  93             icmp_id = None
  94             icmp_seq = 1
  95             count = 0
  96             for p in out:
  97                 ip = p[IP]
  98                 self.assertEqual(ip.version, 4)
  99                 self.assertEqual(ip.flags, 0)
 100                 self.assertEqual(ip.src, self.pg1.local_ip4)
 101                 self.assertEqual(ip.dst, self.pg1.remote_ip4)
 102                 self.assertEqual(ip.proto, 1)
 103                 self.assertEqual(len(ip.options), 0)
 104                 self.assertGreaterEqual(ip.ttl, 254)
 105                 icmp = p[ICMP]
 106                 self.assertEqual(icmp.type, 8)
 107                 self.assertEqual(icmp.code, 0)
 108                 self.assertEqual(icmp.seq, icmp_seq)
 109                 count = count + 1
 110                 if count >= 3:
 111                     icmp_seq = icmp_seq + 1
 112                     count = 0
 113                 if icmp_id is None:
 114                     icmp_id = icmp.id
 115                 else:
 116                     self.assertEqual(icmp.id, icmp_id)
 117         finally:
 118             self.vapi.cli("show error")