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
9 from vpp_object import VppObject
10 from framework import VppTestCase, VppTestRunner
13 class VppLcpPair(VppObject):
14 def __init__(self, test, phy, host):
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)
25 def remove_vpp_config(self):
26 self._test.vapi.cli("test lcp del phy %s host %s" %
27 (self.phy, self.host))
30 return "lcp:%d:%d" % (self.phy.sw_if_index,
31 self.host.sw_if_index)
33 def query_vpp_config(self):
34 pairs = list(self._test.vapi.vpp.details_iter(
35 self._test.vapi.lcp_itf_pair_get))
38 if p.phy_sw_if_index == self.phy.sw_if_index and \
39 p.host_sw_if_index == self.host.sw_if_index:
44 class TestLinuxCP(VppTestCase):
45 """ Linux Control Plane """
47 extra_vpp_plugin_config = ["plugin",
51 "linux_cp_unittest_plugin.so",
56 super(TestLinuxCP, cls).setUpClass()
59 def tearDownClass(cls):
60 super(TestLinuxCP, cls).tearDownClass()
63 super(TestLinuxCP, self).setUp()
65 # create 4 pg interfaces so there are a few addresses
67 self.create_pg_interfaces(range(4))
69 for i in self.pg_interfaces:
73 for i in self.pg_interfaces:
75 super(TestLinuxCP, self).tearDown()
77 def test_linux_cp_tap(self):
84 arp_opts = {"who-has": 1, "is-at": 2}
86 # create two pairs, wihch a bunch of hots on the phys
87 hosts = [self.pg0, self.pg1]
88 phys = [self.pg2, self.pg3]
93 phy.generate_remote_hosts(4)
94 phy.configure_ipv4_neighbors()
96 pair1 = VppLcpPair(self, phys[0], hosts[0]).add_vpp_config()
97 pair2 = VppLcpPair(self, phys[1], hosts[1]).add_vpp_config()
99 self.logger.info(self.vapi.cli("sh lcp adj verbose"))
100 self.logger.info(self.vapi.cli("sh lcp"))
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) /
116 rxs = self.send_and_expect(host, [p], phy)
118 # verify packet is unchanged
120 self.assertEqual(p.show2(True), rx.show2(True))
122 # ARPs x-connect to phy
123 p = (Ether(dst="ff:ff:ff:ff:ff:ff",
124 src=phy.remote_hosts[j].mac) /
126 hwdst=phy.remote_hosts[j].mac,
129 pdst=phy.remote_hosts[j].ip4))
131 rxs = self.send_and_expect(host, [p], phy)
133 # verify packet is unchanged
135 self.assertEqual(p.show2(True), rx.show2(True))
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) /
147 rxs = self.send_and_expect(phy, [p], host)
149 # verify packet is unchanged
151 self.assertEqual(p.show2(True), rx.show2(True))
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) /
157 hwsrc=phy.remote_hosts[j].mac,
160 psrc=phy.remote_hosts[j].ip4))
162 rxs = self.send_and_expect(phy, [p], host)
164 # verify packet is unchanged
166 self.assertEqual(p.show2(True), rx.show2(True))
173 if __name__ == '__main__':
174 unittest.main(testRunner=VppTestRunner)