7 from config import config
8 from framework import VppTestCase, VppTestRunner
9 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
10 from scapy.layers.inet import IP, ICMP
11 from scapy.layers.l2 import Ether
12 from scapy.packet import Raw
15 @unittest.skipUnless(config.gcov, "part of code coverage tests")
16 class TestVlib(VppTestCase):
17 """Vlib Unit Test Cases"""
23 super(TestVlib, cls).setUpClass()
26 def tearDownClass(cls):
27 super(TestVlib, cls).tearDownClass()
30 super(TestVlib, self).setUp()
33 super(TestVlib, self).tearDown()
35 def test_vlib_main_unittest(self):
36 """Vlib main.c Code Coverage Test"""
40 "packet-generator new {\n"
45 " node ethernet-input\n"
47 " IP6: 00:d0:2d:5e:86:85 -> 00:0d:ea:d0:00:00\n"
48 " ICMP: db00::1 -> db00::2\n"
52 "event-logger trace dispatch",
55 "event-logger resize 102400",
56 "event-logger restart",
57 "pcap dispatch trace on max 100 buffer-trace pg-input 15",
59 "show event-log 100 all",
62 "pcap dispatch trace",
63 "pcap dispatch trace status",
64 "pcap dispatch trace off",
65 "show vlib frame-allocation",
69 r = self.vapi.cli_return_response(cmd)
71 if hasattr(r, "reply"):
72 self.logger.info(cmd + " FAIL reply " + r.reply)
74 self.logger.info(cmd + " FAIL retval " + str(r.retval))
76 def test_vlib_node_cli_unittest(self):
77 """Vlib node_cli.c Code Coverage Test"""
81 "packet-generator new {\n"
86 " node ethernet-input\n"
88 " IP6: 00:d0:2d:5e:86:85 -> 00:0d:ea:d0:00:00\n"
89 " ICMP: db00::1 -> db00::2\n"
94 "show vlib graph ethernet-input",
96 "show vlib graphviz graphviz.dot",
98 "show runtime ethernet-input",
99 "show runtime brief verbose max summary",
102 "show node ethernet-input",
103 "show node pg-input",
105 "set node function no-such-node",
106 "set node function cdp-input default",
107 "set node function ethernet-input default",
108 "set node function ethernet-input bozo",
109 "set node function ethernet-input",
114 r = self.vapi.cli_return_response(cmd)
116 if hasattr(r, "reply"):
117 self.logger.info(cmd + " FAIL reply " + r.reply)
119 self.logger.info(cmd + " FAIL retval " + str(r.retval))
121 def test_vlib_buffer_c_unittest(self):
122 """Vlib buffer.c Code Coverage Test"""
126 "packet-generator new {\n"
131 " node ethernet-input\n"
133 " IP6: 00:d0:2d:5e:86:85 -> 00:0d:ea:d0:00:00\n"
134 " ICMP: db00::1 -> db00::2\n"
138 "event-logger trace",
139 "event-logger trace enable",
140 "event-logger trace api cli barrier",
142 "show interface bogus",
143 "event-logger trace disable api cli barrier",
144 "event-logger trace circuit-node ethernet-input",
145 "event-logger trace circuit-node ethernet-input disable",
149 "show memory api-segment stats-segment main-heap verbose",
150 "leak-check { show memory }",
152 "memory-trace main-heap",
153 "memory-trace main-heap api-segment stats-segment",
154 "leak-check { show version }",
156 "comment { show version }",
157 "uncomment { show version }",
158 "show memory main-heap",
161 "test heap-validate",
162 "memory-trace main-heap disable",
170 r = self.vapi.cli_return_response(cmd)
172 if hasattr(r, "reply"):
173 self.logger.info(cmd + " FAIL reply " + r.reply)
175 self.logger.info(cmd + " FAIL retval " + str(r.retval))
177 def test_vlib_format_unittest(self):
178 """Vlib format.c Code Coverage Test"""
182 "classify filter pcap mask l2 proto match l2 proto 0x86dd",
183 "classify filter pcap del",
188 r = self.vapi.cli_return_response(cmd)
190 if hasattr(r, "reply"):
191 self.logger.info(cmd + " FAIL reply " + r.reply)
193 self.logger.info(cmd + " FAIL retval " + str(r.retval))
195 def test_vlib_main_unittest(self):
196 """Private Binary API Segment Test (takes 70 seconds)"""
198 vat_path = config.vpp + "_api_test"
199 vat = pexpect.spawn(vat_path, ["socket-name", self.get_api_sock_path()])
200 vat.expect("vat# ", timeout=10)
201 vat.sendline("sock_init_shm")
202 vat.expect("vat# ", timeout=10)
203 vat.sendline("sh api cli")
204 vat.kill(signal.SIGKILL)
206 self.logger.info("vat terminated, 70 second wait for the Reaper")
208 self.logger.info("Reaper should be complete...")
211 """Fixed-size Pool Test"""
218 r = self.vapi.cli_return_response(cmd)
220 if hasattr(r, "reply"):
221 self.logger.info(cmd + " FAIL reply " + r.reply)
223 self.logger.info(cmd + " FAIL retval " + str(r.retval))
226 class TestVlibFrameLeak(VppTestCase):
227 """Vlib Frame Leak Test Cases"""
233 super(TestVlibFrameLeak, cls).setUpClass()
236 def tearDownClass(cls):
237 super(TestVlibFrameLeak, cls).tearDownClass()
240 super(TestVlibFrameLeak, self).setUp()
241 # create 1 pg interface
242 self.create_pg_interfaces(range(1))
244 for i in self.pg_interfaces:
250 super(TestVlibFrameLeak, self).tearDown()
251 for i in self.pg_interfaces:
255 def test_vlib_mw_refork_frame_leak(self):
256 """Vlib worker thread refork leak test case"""
259 icmp_load = b"\x0a" * 18
261 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
262 / IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
263 / ICMP(id=icmp_id, seq=icmp_seq)
264 / Raw(load=icmp_load)
268 self.pg0.add_stream(pkt)
269 self.pg_enable_capture(self.pg_interfaces)
272 rx = self.pg0.get_capture(1)
274 self.assertEquals(len(rx), 1)
279 self.assertEqual(ether.src, self.pg0.local_mac)
280 self.assertEqual(ether.dst, self.pg0.remote_mac)
282 self.assertEqual(ipv4.src, self.pg0.local_ip4)
283 self.assertEqual(ipv4.dst, self.pg0.remote_ip4)
285 # Save allocated frame count
287 for fs in self.vapi.cli("show vlib frame-allocation").splitlines()[1:]:
293 frame_allocated[key] = alloc
296 _ = self.create_loopback_interfaces(1)
298 # send the same packet
299 self.pg0.add_stream(pkt)
300 self.pg_enable_capture(self.pg_interfaces)
303 rx = self.pg0.get_capture(1)
305 self.assertEquals(len(rx), 1)
310 self.assertEqual(ether.src, self.pg0.local_mac)
311 self.assertEqual(ether.dst, self.pg0.remote_mac)
313 self.assertEqual(ipv4.src, self.pg0.local_ip4)
314 self.assertEqual(ipv4.dst, self.pg0.remote_ip4)
316 # Check that no frame were leaked during refork
317 for fs in self.vapi.cli("show vlib frame-allocation").splitlines()[1:]:
323 self.assertEqual(frame_allocated[key], alloc)
326 if __name__ == "__main__":
327 unittest.main(testRunner=VppTestRunner)