src/plugins/linux-cp/test/test_linux_cp.py

   1 #!/usr/bin/env python3
   2
   3 import unittest
   4
   5 from scapy.layers.inet import IP, UDP
   6 from scapy.layers.inet6 import IPv6, Raw
   7 from scapy.layers.l2 import Ether, ARP, Dot1Q
   8
   9 from vpp_object import VppObject
  10 from framework import VppTestCase, VppTestRunner
  11
  12
  13 class VppLcpPair(VppObject):
  14     def __init__(self, test, phy, host):
  15         self._test = test
  16         self.phy = phy
  17         self.host = host
  18
  19     def add_vpp_config(self):
  20         self._test.vapi.cli("test lcp add phy %s host %s" %
  21                             (self.phy, self.host))
  22         self._test.registry.register(self, self._test.logger)
  23         return self
  24
  25     def remove_vpp_config(self):
  26         self._test.vapi.cli("test lcp del phy %s host %s" %
  27                             (self.phy, self.host))
  28
  29     def object_id(self):
  30         return "lcp:%d:%d" % (self.phy.sw_if_index,
  31                               self.host.sw_if_index)
  32
  33     def query_vpp_config(self):
  34         pairs = list(self._test.vapi.vpp.details_iter(
  35             self._test.vapi.lcp_itf_pair_get))
  36
  37         for p in pairs:
  38             if p.phy_sw_if_index == self.phy.sw_if_index and \
  39                p.host_sw_if_index == self.host.sw_if_index:
  40                 return True
  41         return False
  42
  43
  44 class TestLinuxCP(VppTestCase):
  45     """ Linux Control Plane """
  46
  47     extra_vpp_plugin_config = ["plugin",
  48                                "linux_cp_plugin.so",
  49                                "{", "enable", "}",
  50                                "plugin",
  51                                "linux_cp_unittest_plugin.so",
  52                                "{", "enable", "}"]
  53
  54     @classmethod
  55     def setUpClass(cls):
  56         super(TestLinuxCP, cls).setUpClass()
  57
  58     @classmethod
  59     def tearDownClass(cls):
  60         super(TestLinuxCP, cls).tearDownClass()
  61
  62     def setUp(self):
  63         super(TestLinuxCP, self).setUp()
  64
  65         # create 4 pg interfaces so there are a few addresses
  66         # in the FIB
  67         self.create_pg_interfaces(range(4))
  68
  69         for i in self.pg_interfaces:
  70             i.admin_up()
  71
  72     def tearDown(self):
  73         for i in self.pg_interfaces:
  74             i.admin_down()
  75         super(TestLinuxCP, self).tearDown()
  76
  77     def test_linux_cp_tap(self):
  78         """ Linux CP TAP """
  79
  80         #
  81         # Setup
  82         #
  83
  84         arp_opts = {"who-has": 1, "is-at": 2}
  85
  86         # create two pairs, wihch a bunch of hots on the phys
  87         hosts = [self.pg0, self.pg1]
  88         phys = [self.pg2, self.pg3]
  89         N_HOSTS = 4
  90
  91         for phy in phys:
  92             phy.config_ip4()
  93             phy.generate_remote_hosts(4)
  94             phy.configure_ipv4_neighbors()
  95
  96         pair1 = VppLcpPair(self, phys[0], hosts[0]).add_vpp_config()
  97         pair2 = VppLcpPair(self, phys[1], hosts[1]).add_vpp_config()
  98
  99         self.logger.info(self.vapi.cli("sh lcp adj verbose"))
 100         self.logger.info(self.vapi.cli("sh lcp"))
 101
 102         #
 103         # Traffic Tests
 104         #
 105
 106         # hosts to phys
 107         for phy, host in zip(phys, hosts):
 108             for j in range(N_HOSTS):
 109                 p = (Ether(src=phy.local_mac,
 110                            dst=phy.remote_hosts[j].mac) /
 111                      IP(src=phy.local_ip4,
 112                         dst=phy.remote_hosts[j].ip4) /
 113                      UDP(sport=1234, dport=1234) /
 114                      Raw())
 115
 116                 rxs = self.send_and_expect(host, [p], phy)
 117
 118                 # verify packet is unchanged
 119                 for rx in rxs:
 120                     self.assertEqual(p.show2(True), rx.show2(True))
 121
 122                 # ARPs x-connect to phy
 123                 p = (Ether(dst="ff:ff:ff:ff:ff:ff",
 124                            src=phy.remote_hosts[j].mac) /
 125                      ARP(op="who-has",
 126                          hwdst=phy.remote_hosts[j].mac,
 127                          hwsrc=phy.local_mac,
 128                          psrc=phy.local_ip4,
 129                          pdst=phy.remote_hosts[j].ip4))
 130
 131                 rxs = self.send_and_expect(host, [p], phy)
 132
 133                 # verify packet is unchanged
 134                 for rx in rxs:
 135                     self.assertEqual(p.show2(True), rx.show2(True))
 136
 137         # phy to host
 138         for phy, host in zip(phys, hosts):
 139             for j in range(N_HOSTS):
 140                 p = (Ether(dst=phy.local_mac,
 141                            src=phy.remote_hosts[j].mac) /
 142                      IP(dst=phy.local_ip4,
 143                         src=phy.remote_hosts[j].ip4) /
 144                      UDP(sport=1234, dport=1234) /
 145                      Raw())
 146
 147                 rxs = self.send_and_expect(phy, [p], host)
 148
 149                 # verify packet is unchanged
 150                 for rx in rxs:
 151                     self.assertEqual(p.show2(True), rx.show2(True))
 152
 153                 # ARPs rx'd on the phy are sent to the host
 154                 p = (Ether(dst="ff:ff:ff:ff:ff:ff",
 155                            src=phy.remote_hosts[j].mac) /
 156                      ARP(op="is-at",
 157                          hwsrc=phy.remote_hosts[j].mac,
 158                          hwdst=phy.local_mac,
 159                          pdst=phy.local_ip4,
 160                          psrc=phy.remote_hosts[j].ip4))
 161
 162                 rxs = self.send_and_expect(phy, [p], host)
 163
 164                 # verify packet is unchanged
 165                 for rx in rxs:
 166                     self.assertEqual(p.show2(True), rx.show2(True))
 167
 168         # cleanup
 169         for phy in phys:
 170             phy.unconfig_ip4()
 171
 172
 173 if __name__ == '__main__':
 174     unittest.main(testRunner=VppTestRunner)