test/test_ping.py

   1 from scapy.layers.inet import IP, ICMP
   2
   3 from framework import VppTestCase
   4 from vpp_ip_route import VppIpInterfaceAddress, VppIpRoute, VppRoutePath
   5 from vpp_neighbor import VppNeighbor
   6
   7 """ TestPing is a subclass of  VPPTestCase classes.
   8
   9 Basic test for sanity check of the ping.
  10
  11 """
  12
  13
  14 class TestPing(VppTestCase):
  15     """Ping Test Case"""
  16
  17     @classmethod
  18     def setUpClass(cls):
  19         super(TestPing, cls).setUpClass()
  20         try:
  21             cls.create_pg_interfaces(range(2))
  22             cls.interfaces = list(cls.pg_interfaces)
  23
  24             for i in cls.interfaces:
  25                 i.admin_up()
  26                 i.config_ip4()
  27                 i.config_ip6()
  28                 i.disable_ipv6_ra()
  29                 i.resolve_arp()
  30                 i.resolve_ndp()
  31         except Exception:
  32             super(TestPing, cls).tearDownClass()
  33             raise
  34
  35     @classmethod
  36     def tearDownClass(cls):
  37         super(TestPing, cls).tearDownClass()
  38
  39     def tearDown(self):
  40         super(TestPing, self).tearDown()
  41
  42     def show_commands_at_teardown(self):
  43         self.logger.info(self.vapi.cli("show hardware"))
  44
  45     def verify_ping_request(self, p, src, dst, seq):
  46         ip = p[IP]
  47         self.assertEqual(ip.version, 4)
  48         self.assertEqual(ip.flags, 0)
  49         self.assertEqual(ip.src, src)
  50         self.assertEqual(ip.dst, dst)
  51         self.assertEqual(ip.proto, 1)
  52         self.assertEqual(len(ip.options), 0)
  53         self.assertGreaterEqual(ip.ttl, 254)
  54         icmp = p[ICMP]
  55         self.assertEqual(icmp.type, 8)
  56         self.assertEqual(icmp.code, 0)
  57         self.assertEqual(icmp.seq, seq)
  58         return icmp
  59
  60     def test_ping_basic(self):
  61         """basic ping test"""
  62         try:
  63             self.pg_enable_capture(self.pg_interfaces)
  64             self.pg_start()
  65             self.logger.info(self.vapi.cli("show ip4 neighbors"))
  66             self.logger.info(self.vapi.cli("show ip6 neighbors"))
  67
  68             remote_ip4 = self.pg1.remote_ip4
  69             ping_cmd = "ping " + remote_ip4 + " interval 0.01 repeat 10"
  70             ret = self.vapi.cli(ping_cmd)
  71             self.logger.info(ret)
  72             out = self.pg1.get_capture(10)
  73             icmp_id = None
  74             icmp_seq = 1
  75             for p in out:
  76                 icmp = self.verify_ping_request(
  77                     p, self.pg1.local_ip4, self.pg1.remote_ip4, icmp_seq
  78                 )
  79                 icmp_seq = icmp_seq + 1
  80                 if icmp_id is None:
  81                     icmp_id = icmp.id
  82                 else:
  83                     self.assertEqual(icmp.id, icmp_id)
  84         finally:
  85             self.vapi.cli("show error")
  86
  87     def test_ping_burst(self):
  88         """burst ping test"""
  89         try:
  90             self.pg_enable_capture(self.pg_interfaces)
  91             self.pg_start()
  92             self.logger.info(self.vapi.cli("show ip neighbors"))
  93
  94             remote_ip4 = self.pg1.remote_ip4
  95             ping_cmd = "ping " + remote_ip4 + " interval 0.01 burst 3"
  96             ret = self.vapi.cli(ping_cmd)
  97             self.logger.info(ret)
  98             out = self.pg1.get_capture(3 * 5)
  99             icmp_id = None
 100             icmp_seq = 1
 101             count = 0
 102             for p in out:
 103                 icmp = self.verify_ping_request(
 104                     p, self.pg1.local_ip4, self.pg1.remote_ip4, icmp_seq
 105                 )
 106                 count = count + 1
 107                 if count >= 3:
 108                     icmp_seq = icmp_seq + 1
 109                     count = 0
 110                 if icmp_id is None:
 111                     icmp_id = icmp.id
 112                 else:
 113                     self.assertEqual(icmp.id, icmp_id)
 114         finally:
 115             self.vapi.cli("show error")
 116
 117     def test_ping_src(self):
 118         """ping with source address set"""
 119
 120         self.pg_enable_capture(self.pg_interfaces)
 121         self.pg_start()
 122         self.logger.info(self.vapi.cli("show ip4 neighbors"))
 123         self.logger.info(self.vapi.cli("show ip6 neighbors"))
 124
 125         nbr_addr = "10.0.0.2"
 126         VppIpInterfaceAddress(self, self.pg1, "10.0.0.1", 24).add_vpp_config()
 127         VppNeighbor(
 128             self, self.pg1.sw_if_index, "00:11:22:33:44:55", nbr_addr
 129         ).add_vpp_config()
 130
 131         ping_cmd = "ping %s interval 0.01 repeat 3" % self.pg1.remote_ip4
 132         ret = self.vapi.cli(ping_cmd)
 133         out = self.pg1.get_capture(3)
 134         icmp_seq = 1
 135         for p in out:
 136             icmp = self.verify_ping_request(
 137                 p, self.pg1.local_ip4, self.pg1.remote_ip4, icmp_seq
 138             )
 139             icmp_seq = icmp_seq + 1
 140
 141         self.pg_enable_capture(self.pg_interfaces)
 142         self.pg_start()
 143         ping_cmd = "ping %s interval 0.01 repeat 3" % nbr_addr
 144         ret = self.vapi.cli(ping_cmd)
 145         out = self.pg1.get_capture(3)
 146         icmp_seq = 1
 147         for p in out:
 148             icmp = self.verify_ping_request(p, "10.0.0.1", nbr_addr, icmp_seq)
 149             icmp_seq = icmp_seq + 1
 150
 151     def test_ping_fib_routed_dst(self):
 152         """ping destination routed according to FIB table"""
 153
 154         try:
 155             self.pg1.generate_remote_hosts(1)
 156             self.pg_enable_capture(self.pg_interfaces)
 157             self.pg_start()
 158             routed_dst = "10.0.2.0"
 159             self.logger.info(self.vapi.cli("show ip4 neighbors"))
 160             VppIpRoute(
 161                 self,
 162                 routed_dst,
 163                 24,
 164                 [VppRoutePath(self.pg1.remote_hosts[0].ip4, self.pg1.sw_if_index)],
 165             ).add_vpp_config()
 166             ping_cmd = "ping %s interval 0.01 repeat 3" % routed_dst
 167             ret = self.vapi.cli(ping_cmd)
 168             self.logger.info(ret)
 169             out = self.pg1.get_capture(3)
 170             icmp_seq = 1
 171             for p in out:
 172                 self.verify_ping_request(p, self.pg1.local_ip4, routed_dst, icmp_seq)
 173                 icmp_seq = icmp_seq + 1
 174         finally:
 175             self.vapi.cli("show error")
 176
 177     def test_ping_api(self):
 178         """ping api"""
 179
 180         try:
 181             self.pg_enable_capture(self.pg_interfaces)
 182             self.pg_start()
 183
 184             ret = self.vapi.want_ping_finished_events(
 185                 address=self.pg1.remote_ip4,
 186                 repeat=4,
 187                 interval=0.2,
 188             )
 189             self.logger.info(ret)
 190             timeout = 1
 191
 192             ev = self.vapi.wait_for_event(timeout, "ping_finished_event")
 193             self.logger.info(ev)
 194             self.assertEqual(ev.request_count, 4)
 195
 196             out = self.pg1.get_capture(4)
 197             icmp_id = None
 198             icmp_seq = 1
 199             for p in out:
 200                 icmp = self.verify_ping_request(
 201                     p, self.pg1.local_ip4, self.pg1.remote_ip4, icmp_seq
 202                 )
 203                 icmp_seq = icmp_seq + 1
 204                 if icmp_id is None:
 205                     icmp_id = icmp.id
 206                 else:
 207                     self.assertEqual(icmp.id, icmp_id)
 208         finally:
 209             self.vapi.cli("show error")